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.models import Comment, 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 = 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("/") @self.limiter.exempt def index() -> str: return flask.render_template("index.html") @self.app.route("/api/comments", methods=["GET"]) async def get_comments() -> list: return await Comment.all().order_by("created_at").values() @self.app.route("/api/comments", methods=["POST"]) async def post_comment() -> tuple[str, int]: data = flask.request.get_json() await Comment.create(content=data["content"]) return ("", 204) @self.app.route("/api/lists//tasks", methods=["GET"]) async def get_tasks(list_name: str) -> list: return await Task.filter(list_name=list_name).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 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) 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