feat: base flask x gunicorn server
Python Lint CI / ruff-format-check (push) Successful in 1m21s
Python Lint CI / ruff (push) Successful in 1m21s
Python Lint CI / ty (push) Successful in 1m29s

This commit is contained in:
2026-06-30 17:09:16 +02:00
parent e19a9cfdcc
commit 83c8e4415b
8 changed files with 299 additions and 46 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
import importlib.metadata
PKG_NAME = __name__
PKG_NAME = "tout-doux"
PKG_VERSION = importlib.metadata.version(PKG_NAME)
+5 -3
View File
@@ -1,13 +1,15 @@
import logging
import sys
from . import PKG_NAME, PKG_VERSION
from .logs import setup_logs
from .params import parse_parameters
from app import PKG_NAME, PKG_VERSION
from app.logs import setup_logs
from app.params import parse_parameters
from app.server import Server
def main() -> int:
params = parse_parameters(sys.argv[1:])
setup_logs(params)
logging.getLogger().info("%s %s", PKG_NAME, PKG_VERSION)
Server(params).register_routes().start()
return 0
+32 -1
View File
@@ -2,8 +2,11 @@ import enum
import logging
import typing
import gunicorn.config
import gunicorn.glogging
if typing.TYPE_CHECKING:
from .params import Parameters
from app.params import Parameters
class TermColor(enum.StrEnum):
@@ -57,6 +60,20 @@ class ColoredLoggingFormatter(logging.Formatter):
self.trace: bool = trace
super().__init__()
@classmethod
def flat_format(cls, level: int) -> str:
return (
cls.pre_format
+ cls.FORMAT_COLORS.get(
level,
TermColor.MAGENTA,
)
+ TermColor.BOLD
+ cls.level_format
+ TermColor.RESET
+ cls.post_format
)
@typing.override
def format(self, record: logging.LogRecord) -> str:
log_color: TermColor | str = self.FORMAT_COLORS.get(
@@ -75,6 +92,20 @@ class ColoredLoggingFormatter(logging.Formatter):
return formatter.format(record)
class GunicornLogger(gunicorn.glogging.Logger):
@typing.override
def __init__(self, cfg: gunicorn.config.Config) -> None:
super().__init__(cfg)
self._set_handler(
self.error_log, cfg.errorlog, ColoredLoggingFormatter(trace=False)
)
if cfg.accesslog is not None:
self._set_handler(
self.access_log, cfg.accesslog, ColoredLoggingFormatter(trace=False)
)
def setup_logs(params: Parameters) -> None:
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(ColoredLoggingFormatter(trace=params.debug))
+86 -1
View File
@@ -1,20 +1,70 @@
import argparse
import dataclasses
import os
import typing
from . import PKG_NAME
from app import PKG_NAME
@dataclasses.dataclass(frozen=True, slots=True)
class Parameters:
debug: bool = False
quiet: bool = False
env: str = "production"
bind: str = "0.0.0.0"
port: int = 5000
workers: int = 4
timeout: int = 120
@classmethod
def from_namespace(cls, args: argparse.Namespace) -> Parameters:
return Parameters(**vars(args))
def __get_env_str(var: str, default: str) -> str:
if (result := os.getenv(var)) is None:
return default
return result
def __get_env_int(var: str, default: int) -> int:
value = __get_env_str(var, str(default))
if value.isdecimal():
return int(value)
return default
def __add_arg_str(
parser: argparse.ArgumentParser,
*flags: str,
env_var: str,
default: str,
help_txt: str,
) -> None:
parser.add_argument(
*flags,
metavar=env_var,
default=__get_env_str(env_var, default),
help=f"{help_txt} (default: {default})" if len(default) else help_txt,
)
def __add_arg_int(
parser: argparse.ArgumentParser,
*flags: str,
env_var: str,
default: int,
help_txt: str,
) -> None:
parser.add_argument(
*flags,
type=int,
metavar=env_var,
default=__get_env_int(env_var, default),
help=f"{help_txt} (default: {default})",
)
def parse_parameters(args: typing.Sequence[str]) -> Parameters:
default_values = Parameters()
parser = argparse.ArgumentParser(
@@ -27,5 +77,40 @@ def parse_parameters(args: typing.Sequence[str]) -> Parameters:
parser.add_argument(
"--quiet", action=argparse.BooleanOptionalAction, default=default_values.quiet
)
__add_arg_str(
parser,
"--env",
env_var="APP_ENV",
default=default_values.env,
help_txt="environnement",
)
__add_arg_str(
parser,
"--bind",
env_var="BIND",
default=default_values.bind,
help_txt="server bind address",
)
__add_arg_int(
parser,
"--port",
env_var="PORT",
default=default_values.port,
help_txt="server http port",
)
__add_arg_int(
parser,
"--workers",
env_var="WORKERS",
default=default_values.workers,
help_txt="WSGI workers",
)
__add_arg_int(
parser,
"--timeout",
env_var="TIMEOUT",
default=default_values.timeout,
help_txt="WSGI timeout",
)
parsed_args = parser.parse_args(args)
return Parameters.from_namespace(parsed_args)
+49
View File
@@ -0,0 +1,49 @@
import typing
import flask
import gunicorn.app.base
from app import PKG_NAME
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)
super().__init__()
def register_routes(self) -> typing.Self:
@self.app.route("/")
def hello_world() -> str:
return "<p>Hello, World!</p>"
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
)