122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
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/<list_name>", methods=["GET"])
|
|
async def get_tasks(list_name: str) -> list:
|
|
return await Task.filter(list=list_name).values()
|
|
|
|
@self.app.route("/api/lists/<list_name>", methods=["POST"])
|
|
async def post_task(list_name: str) -> typing.Any:
|
|
data = flask.request.get_json()
|
|
data["list"] = list_name
|
|
parsed_data = Task.validate_create(data)
|
|
if parsed_data is None:
|
|
flask.abort(400)
|
|
return (await Task.create(**parsed_data)).get().values()
|
|
|
|
@self.app.route("/api/lists/<list_name>/task/<task_uuid>", methods=["PUT"])
|
|
async def put_task(list_name: str, task_uuid: str) -> typing.Any:
|
|
data = flask.request.get_json()
|
|
data["list"] = list_name
|
|
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.get().values()
|
|
except tortoise.exceptions.ObjectDoesNotExistError:
|
|
flask.abort(404)
|
|
|
|
@self.app.route("/api/lists/<list_name>/task/<task_uuid>", methods=["DELETE"])
|
|
async def delete_task(list_name: str, task_uuid: str) -> typing.Any:
|
|
data = flask.request.get_json()
|
|
data["list"] = list_name
|
|
parsed_data = Task.validate_update(data)
|
|
if parsed_data is None:
|
|
flask.abort(400)
|
|
try:
|
|
task = await Task.get(id=task_uuid)
|
|
await task.delete()
|
|
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
|
|
|
|
def start(self) -> None:
|
|
if self.env == "production":
|
|
self.run()
|
|
else:
|
|
self.app.run(
|
|
host=self.bind,
|
|
port=self.port,
|
|
debug=self.debug,
|
|
load_dotenv=False,
|
|
use_evalex=False,
|
|
use_reloader=False,
|
|
)
|