import logging import typing import flask import flask_limiter import flask_limiter.util import gunicorn.app.base import tortoise 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("/") @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/lists//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//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/", 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/", 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