""" Database management """ from datetime import datetime from urllib.parse import urlparse, parse_qsl from pony.orm import Database, Required, Optional, Set, PrimaryKey, LongStr db = Database() class Service(db.Entity): id = PrimaryKey(int, auto=True) name = Required(str, unique=True) localisation = Optional(str) max_appointement = Required(int) plannings = Set('Planning') service_category = Required('Service_category') appointments = Set('Appointment') description = Optional(str) class Planning(db.Entity): id = PrimaryKey(int, auto=True) name = Required(str, unique=True) working_plan = Required(LongStr) services = Set(Service) appointments = Set('Appointment') class Setting(db.Entity): id = PrimaryKey(int, auto=True) name = Required(str, unique=True) value = Optional(str) class Appointment(db.Entity): id = PrimaryKey(int, auto=True) book_datetime = Required(datetime) start_datetime = Required(datetime) end_datetime = Required(str) localisation = Optional(str, nullable=True) is_unavaillable = Required(bool) service = Required(Service) planning = Required(Planning) ticket = Optional('Ticket') user = Optional('User') notes = Optional(LongStr) class User(db.Entity): id = PrimaryKey(int, auto=True) first_name = Required(str) last_name = Required(str) email = Required(str, unique=True) localisation = Optional(str) mobile_number = Optional(str) phone_number = Optional(str) tickets = Set('Ticket') appointments = Set(Appointment) notes = Optional(str) class Ticket(db.Entity): id = PrimaryKey(int, auto=True) ref = Optional(str, unique=True) service_category = Required('Service_category') user = Required(User) appointment = Required(Appointment) notes = Optional(str) class Service_category(db.Entity): id = PrimaryKey(int, auto=True) name = Required(str) tickets = Set(Ticket) duration = Required(int) free_order = Required(bool) services = Set(Service) description = Optional(str) def parse_uri(database_uri): if not isinstance(database_uri, str): raise TypeError("Expecting a string") uri = urlparse(database_uri) args = dict(parse_qsl(uri.query)) if uri.port is not None: args["port"] = uri.port if uri.scheme == "sqlite": path = uri.path if not path: path = ":memory:" elif path[0] == "/": path = path[1:] return dict(provider="sqlite", filename=path, create_db=True, **args) elif uri.scheme in ("postgres", "postgresql"): return dict( provider="postgres", user=uri.username, password=uri.password, host=uri.hostname, dbname=uri.path[1:], **args ) elif uri.scheme == "mysql": args.setdefault("charset", "utf8mb4") args.setdefault("binary_prefix", True) return dict( provider="mysql", user=uri.username, passwd=uri.password, host=uri.hostname, db=uri.path[1:], **args ) return dict() def init_database(database_uri): settings = parse_uri(database_uri) db.bind(**settings) db.generate_mapping(create_tables=True) def release_database(): db.disconnect()