tout-doux/app/server.py

import logging
import typing

import flask
import flask_limiter
import flask_limiter.util
import gunicorn.app.base
import tortoise
from tortoise.functions import Count

from app import PKG_NAME
from app.lang import get_lang
from app.models import Task

if typing.TYPE_CHECKING:
    from app.params import Parameters


class Server(gunicorn.app.base.BaseApplication):
    def __init__(self, params: Parameters) -> None:
        self.env = params.env
        self.bind = params.bind
        self.port = params.port
        self.debug = params.debug
        self.workers = params.workers
        self.timeout = params.timeout
        self.app_url = params.app_url
        self.app = flask.Flask(
            PKG_NAME, static_folder="dist/assets", template_folder="dist"
        )
        self.limiter = flask_limiter.Limiter(
            flask_limiter.util.get_remote_address,
            app=self.app,
            default_limits=[params.api_rate_limit],
            storage_uri=params.cache_uri,
        )
        self._index_content: str | None = None
        self.logger: logging.Logger = logging.getLogger(self.__class__.__name__)
        super().__init__()

    def register_routes(self) -> typing.Self:
        @self.app.route("/", defaults={"path": ""})
        @self.app.route("/<path>")
        @self.limiter.exempt
        def index(path: str) -> str:
            locale = "en"
            if "lang" in flask.request.cookies:
                locale = flask.request.cookies["lang"]
            if "lang" in flask.request.args:
                locale = flask.request.args["lang"]
            return flask.render_template(
                "index.html",
                title=("Tout Doux" if not len(path) else f"Tout Doux - {path}"),
                app_url=self.app_url,
                path=path,
                locale=locale,
                lang=lambda key: get_lang(locale, key),
            )

        @self.app.route("/api/meta", methods=["GET"])
        async def metadata() -> dict:
            return {
                "tasks": await Task.all().count(),
                "lists": (
                    await Task.all()
                    .annotate(count=Count("list_name", distinct=True))
                    .first()
                    .values("count")
                )["count"],
            }

        @self.app.route("/api/lists/<list_name>/tasks", methods=["GET"])
        async def get_tasks(list_name: str) -> list:
            return await Task.filter(list_name=list_name[:128]).values()

        @self.app.route("/api/lists/<list_name>/tasks", methods=["POST"])
        async def post_task(list_name: str) -> typing.Any:
            data = flask.request.get_json()
            logging.getLogger(self.__class__.__name__).info("%s", data)
            data["list_name"] = list_name[:128]
            parsed_data = Task.validate_create(data)
            if parsed_data is None:
                flask.abort(400)
            return (await Task.create(**parsed_data)).serialize()

        @self.app.route("/api/tasks/<task_uuid>", methods=["PUT"])
        async def put_task(task_uuid: str) -> typing.Any:
            data = flask.request.get_json()
            parsed_data = Task.validate_update(data)
            if parsed_data is None:
                flask.abort(400)
            try:
                task = await Task.get(id=task_uuid)
                await task.update_from_dict(parsed_data)
                await task.save()
                return task.serialize()
            except tortoise.exceptions.ValidationError:
                flask.abort(400)
            except tortoise.exceptions.ObjectDoesNotExistError:
                flask.abort(404)

        @self.app.route("/api/tasks/<task_uuid>", methods=["DELETE"])
        async def delete_task(task_uuid: str) -> typing.Any:
            try:
                task = await Task.get(id=task_uuid)
                await task.delete()
            except tortoise.exceptions.ValidationError:
                flask.abort(400)
            except tortoise.exceptions.ObjectDoesNotExistError:
                flask.abort(404)
            return ""

        return self

    @typing.override
    def load_config(self) -> None:
        if self.cfg is not None:
            self.cfg.set("bind", f"{self.bind}:{self.port}")
            self.cfg.set("workers", self.workers)
            self.cfg.set("timeout", self.timeout)
            self.cfg.set("logger_class", "app.logs.GunicornLogger")
            self.cfg.set("errorlog", "-")
            self.cfg.set("accesslog", "-")

    def load(self) -> flask.Flask:
        return self.app