CyberCoffee API Rest (With Falcon and Peewee)

cybercoffee.py

import json

import falcon
import falcon.asgi
import peewee

DATABASE_FILENAME = "cybercoffee.db"


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
        return cls._instances[cls]


class Database(peewee.SqliteDatabase, metaclass=Singleton):
    def __init__(self, *args, **kwargs):
        super().__init__(DATABASE_FILENAME, *args, **kwargs)
        self.__all_model_db = set()

    def attach_model(self, cls):
        self.__all_model_db.add(cls)
        return cls

    def connect(self, reuse_if_open=False):
        super().connect()

        self.create_tables(list(self.__all_model_db))


db = Database()


class BaseModel(peewee.Model):
    class Meta:
        database = db

    @staticmethod
    def all_list():
        return []


@db.attach_model
class Clients(BaseModel):
    memid = peewee.AutoField(unique=True)
    username = peewee.CharField(null=True, unique=True)
    firstname = peewee.CharField(null=False, unique=True)
    lastname = peewee.CharField(null=False, unique=True)
    address = peewee.CharField(max_length=300, null=True)
    email = peewee.CharField(null=False)
    telephone = peewee.CharField(null=False)
    zipcode = peewee.IntegerField(null=True)


@db.attach_model
class VideoGameConsole(BaseModel):
    memid = peewee.AutoField(unique=True)
    model = peewee.CharField(null=False, unique=True)
    year = peewee.IntegerField(null=False, unique=True)
    brand = peewee.CharField(null=False, unique=True)
    mac_address = peewee.IntegerField(null=False, unique=True)
    status = peewee.CharField(null=True)
    comment = peewee.CharField(null=True)


@db.attach_model
class VideoGame(BaseModel):
    memid = peewee.AutoField(unique=True)
    name = peewee.CharField(null=False, unique=True)
    release = peewee.DateField(null=False, unique=True)
    platform = peewee.ForeignKeyField(VideoGameConsole, backref="model")


@db.attach_model
class VideoGameReservation(BaseModel):
    memid = peewee.AutoField(unique=True)
    date = peewee.DateField(null=False, unique=True)
    client = peewee.ForeignKeyField(Clients, backref="memid")
    videogame = peewee.ForeignKeyField(VideoGame, backref="name")


@db.attach_model
class GameSession(BaseModel):
    memid = peewee.AutoField(unique=True)

    date_init = peewee.DateField(null=False, unique=True)
    date_end = peewee.DateField(null=False)

    console = peewee.ForeignKeyField(VideoGameConsole, backref="memid")
    reservation = peewee.ForeignKeyField(VideoGameReservation, backref="memid")


class NotFoundHandler(metaclass=Singleton):
    async def on_get(self, req, resp, *args, **kwargs):
        resp.status = falcon.HTTP_404
        resp.body = 'No found'


async def json_stream(bounded_stream: falcon.BoundedStream):
    return json.loads(await bounded_stream.read())


class ClientResource(metaclass=Singleton):
    async def on_post(self, req: falcon.Request, resp: falcon.Response):
        try:
            data = await json_stream(req.bounded_stream)

            Clients.create(
                firstname=data["firstname"],
                lastname=data["lastname"],
                email=data["email"],
                telephone=data["telephone"]
            )

            resp.media = {"message": f"Success"}
        except Exception as e:
            resp.status = falcon.HTTP_500
            resp.media = {"message": f"Failure", "exception": str(e)}

    async def on_get(self, req: falcon.Request, resp: falcon.Response):
        resp.status = falcon.HTTP_200
        resp.media = list(Clients.select().dicts())


class VideoGamesResource(metaclass=Singleton):
    async def on_post(self, req: falcon.Request, resp: falcon.Response):
        try:
            data = await json_stream(req.bounded_stream)

            VideoGameConsole.create(
                model=data["model"],
                year=int(data["year"]),
                brand=data["brand"],
                mac_address=data["mac_address"]
            )

            resp.media = {"message": f"Success"}
        except Exception as e:
            resp.status = falcon.HTTP_500
            resp.media = {"message": f"Failure", "exception": str(e)}

    async def on_get(self, req: falcon.Request, resp: falcon.Response):
        resp.status = falcon.HTTP_200
        resp.media = list(VideoGameConsole.select().dicts())


db.connect()

app = falcon.asgi.App()
app.add_route('/clients', ClientResource())
app.add_route('/videogames', VideoGamesResource())
app.add_error_handler(falcon.HTTPNotFound, NotFoundHandler().on_get)