Files
tout-doux/app/server.py
T
klemek 4e7f0476e0
Python Lint CI / ruff-format-check (push) Successful in 54s
Python Lint CI / ty (push) Successful in 54s
Python Lint CI / ruff (push) Successful in 54s
feat: configurable API rate limit
2026-07-03 15:41:57 +02:00

80 lines
2.4 KiB
Python

import logging
import typing
import flask
import flask_limiter
import flask_limiter.util
import gunicorn.app.base
from app import PKG_NAME
from app.models.comment import Comment
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)
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,
)