146 lines
3.7 KiB
Python
146 lines
3.7 KiB
Python
""" 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(datetime)
|
|
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)
|
|
planner = Optional("Planner")
|
|
|
|
|
|
class Ticket(db.Entity):
|
|
id = PrimaryKey(int, auto=True)
|
|
ref = Optional(str, unique=True)
|
|
service_category = Required("Service_category")
|
|
customer = Required(Customer)
|
|
appointment = Optional(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(db.Entity):
|
|
plannings = Set(Planning)
|
|
password = Required(str)
|
|
is_admin = Optional(bool)
|
|
customer = Required(Customer)
|
|
|
|
|
|
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()
|
|
db.provider = None
|
|
db.schema = None
|