""" Database management """ from datetime import datetime from urllib.parse import urlparse, parse_qsl from pony.orm import Database, Required, Optional, Set, PrimaryKey, LongStr, Json from pony.orm import DatabaseError 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(Json) services = Set(Service) appointments = Set("Appointment") planners = Set("Planner") 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") customer = Optional("Customer") notes = Optional(LongStr) class Customer(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") customer = Required(Customer) 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) class Planner(Customer): plannings = Set(Planning) password = Required(str) is_admin = Optional(bool) 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(check_tables=False) try: db.check_tables() except DatabaseError: db.create_tables() def release_database(): db.disconnect()