feat(optimize): optimize project template

.env_example

@@ -7,6 +7,10 @@ SQLALCHEMY_DATABASE_URI=postgresql+asyncpg://demo:91fb8e9e009f5b9ce1854d947e6fe4
 REDIS_DSN=redis://:cfe1c2c4703abb205d71abdc07cc3f3d@localhost:6379
 
 APP_ENV=PROD
+RUNNING_MODE=uvicorn
+WORKERS=1
+LISTENING_HOST=0.0.0.0
+LISTENING_PORT=8000
 
 # docker compose
 DEFAULT_DB_PASSWORD=91fb8e9e009f5b9ce1854d947e6fe4a3

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
     "sqladmin>=0.16.1",
     "cryptography>=42.0.7",
     "bcrypt>=4.1.3",
+    "gunicorn>=22.0.0",
 ]
 readme = "README.md"
 requires-python = ">= 3.12"
@@ -60,7 +61,7 @@ target-version = "py312"
 
 [tool.ruff.lint]
 select = ["ALL"]
-ignore = ["D", "G002", "DTZ003", "ANN401", "ANN101", "ANN102", "EM101", "PD901", "COM812", "ISC001", "FBT", "A003", "PLR0913", "G004"]
+ignore = ["D", "G002", "DTZ003", "ANN401", "ANN101", "ANN102", "EM101", "PD901", "COM812", "ISC001", "FBT", "A003", "PLR0913", "G004", "E501"]
 fixable = ["ALL"]
 
 
@@ -69,7 +70,7 @@ fixable = ["ALL"]
 "tests/*.py" = ["S101", "ANN201"]
 "*exceptions.py" = ["ARG001"]
 "models.py" = ["RUF012"]
-"restapi.py" = ["A002", "B008"]
+"api.py" = ["A002", "B008"]
 "deps.py" = ["B008"]
 "src/internal/api.py" = ["ARG001"]
 "src/auth/schemas.py" = ["N815"] # frontend menu

requirements-dev.lock

@@ -48,6 +48,8 @@ filelock==3.13.1
     # via virtualenv
 greenlet==3.0.3
     # via sqlalchemy
+gunicorn==22.0.0
+    # via fastapi-enterprise-template
 h11==0.14.0
     # via httpcore
     # via uvicorn
@@ -82,6 +84,7 @@ numpy==1.26.3
     # via pandas
 packaging==23.2
     # via black
+    # via gunicorn
     # via pytest
 pandas==2.1.4
     # via fastapi-enterprise-template

requirements.lock

@@ -34,6 +34,8 @@ fastapi==0.108.0
     # via fastapi-enterprise-template
 greenlet==3.0.3
     # via sqlalchemy
+gunicorn==22.0.0
+    # via fastapi-enterprise-template
 h11==0.14.0
     # via httpcore
     # via uvicorn
@@ -56,6 +58,8 @@ markupsafe==2.1.3
     # via wtforms
 numpy==1.26.3
     # via pandas
+packaging==24.1
+    # via gunicorn
 pandas==2.1.4
     # via fastapi-enterprise-template
 phonenumbers==8.13.27

src/__main__.py

@@ -0,0 +1,65 @@
+from typing import Any
+
+from fastapi import FastAPI
+
+from src.app import app
+from src.core.config import settings
+from src.loggers import LOGGING, configure_logger
+
+
+def gunicorn_options() -> dict[str, Any]:
+    from gunicorn import glogging
+
+    class GunicornLogger(glogging.Logger):
+        def setup(self, cfg: Any) -> None:  # noqa: ARG002
+            configure_logger()
+
+    return {
+        "bind": f"{settings.LISTENING_HOST}:{settings.LISTENING_PORT}",
+        "workers": settings.WORKERS,
+        "worker_class": "uvicorn.workers.UvicornWorker",
+        "preload": "-",
+        "forwarded_allow_ips": "*",
+        "accesslog": "-",
+        "errorlog": "-",
+        "logger_class": GunicornLogger,
+    }
+
+
+if __name__ == "__main__":
+    if settings.RUNNING_MODE == "uvicorn":
+        import uvicorn
+
+        uvicorn.run(
+            app,
+            host=settings.LISTENING_HOST,
+            port=settings.LISTENING_PORT,
+            log_config=LOGGING,
+            proxy_headers=True,
+            forwarded_allow_ips="*",
+            loop="uvloop",
+            http="httptools",
+        )
+
+    else:
+        from gunicorn.app.base import BaseApplication
+
+        class StandaloneApplication(BaseApplication):
+            def __init__(self, app: FastAPI, options: dict | None = None) -> None:
+                self.options = options or {}
+                self.application: FastAPI = app
+                super().__init__()
+
+            def load_config(self) -> None:
+                assert self.cfg is not None  # noqa: S101
+                # Filter out options that are not recognized by gunicorn
+                filtered_options = {key: value for key, value in self.options.items() if key in self.cfg.settings}
+
+                # Set the filtered options
+                for key, value in filtered_options.items():
+                    self.cfg.set(key.lower(), value)
+
+            def load(self) -> FastAPI:
+                return self.application
+
+        StandaloneApplication(app, gunicorn_options()).run()

src/app.py

@@ -4,13 +4,14 @@
 import redis.asyncio as aioreids
 import sentry_sdk
 from fastapi import FastAPI
+from fastapi.responses import HTMLResponse
 from starlette.middleware.cors import CORSMiddleware
 from starlette.middleware.errors import ServerErrorMiddleware
 
 from src.core.config import _Env, settings
 from src.core.errors.auth_exceptions import default_exception_handler, exception_handlers, sentry_ignore_errors
 from src.libs.redis import cache
-from src.openapi import openapi_description
+from src.openapi import get_open_api_intro, get_stoplight_elements_html
 from src.register.middlewares import RequestMiddleware
 from src.register.routers import router
 
@@ -38,9 +39,31 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:  # noqa: ARG001
         title=settings.PROJECT_NAME,
         version=settings.VERSION,
         summary=settings.DESCRIPTION,
-        description=openapi_description,
+        description=get_open_api_intro(),
         lifespan=lifespan,
+        docs_url="/api/docs",
+        redoc_url="/api/redoc",
+        openapi_url="/api/openapi.json",
     )
+
+    @app.get(
+        "/api/health", include_in_schema=False, tags=["Internal"], operation_id="e7372032-61c5-4e3d-b2f1-b788fe1c52ba"
+    )
+    def health() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.get(
+        "/api/version", include_in_schema=False, tags=["Internal"], operation_id="47918987-15d9-4eea-8c29-e73cb009a4d5"
+    )
+    def version() -> dict[str, str]:
+        return {"version": settings.VERSION}
+
+    @app.get(
+        "/api/elements", include_in_schema=False, tags=["Docs"], operation_id="1a4987dd-6c38-4502-a879-3fe35050ae38"
+    )
+    def get_stoplight_elements() -> HTMLResponse:
+        return get_stoplight_elements_html(openapi_url="/api/openapi.json", title=settings.PROJECT_NAME)
+
     app.include_router(router, prefix="/api")
     for handler in exception_handlers:
         app.add_exception_handler(exc_class_or_status_code=handler["exception"], handler=handler["handler"])

src/core/config.py

@@ -1,6 +1,7 @@
 import tomllib
 from enum import StrEnum
 from pathlib import Path
+from typing import Literal
 
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -44,6 +45,10 @@ class Settings(BaseSettings):
     REDIS_DSN: str = Field(default="redis://:cfe1c2c4703abb205d71abdc07cc3f3d@localhost:6379")
 
     ENV: str = _Env.DEV.name
+    RUNNING_MODE: Literal["uvicorn", "gunicorn"] | None = Field(default="uvicorn")
+    WORKERS: int | None = Field(default=1, gt=0)
+    LISTENING_HOST: str = Field(default="0.0.0.0")  # noqa: S104
+    LISTENING_PORT: int = Field(default=8000, gt=0, le=65535)
 
     model_config = SettingsConfigDict(env_file=f"{PROJECT_DIR}/.env", case_sensitive=True, extra="allow")
 

src/core/utils/i18n.py

@@ -6,7 +6,7 @@
 
 from src.core.utils.context import locale_ctx
 from src.core.utils.singleton import singleton
-from src.openapi import translations
+from src.core.utils.translations import translations
 
 
 @singleton
@@ -36,5 +36,5 @@ def _find(self, language: str, path: str) -> dict | str:
             return f"missing translation for {language}"
 
 
-_i18n: I18n = I18n()
+_i18n = I18n()
 _: Callable[..., dict | str] = _i18n.gettext

src/core/utils/processors.py

@@ -0,0 +1,45 @@
+import io
+from pathlib import Path
+
+import pandas as pd
+
+
+def bytes_to_human_readable(bytes_: int, format_string: str = "%(value).1f%(symbol)s") -> str:
+    symbols = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
+    prefix = {}
+    for i, s in enumerate(symbols[1:]):
+        prefix[s] = 1 << (i + 1) * 10
+    for symbol in reversed(symbols[1:]):
+        if bytes_ >= prefix[symbol]:
+            value = float(bytes_) / prefix[symbol]
+            return format_string % locals()
+    return format_string % {"symbol": symbols[0], "value": bytes_}
+
+
+def export_csv(data: list[dict]) -> str:
+    df = pd.DataFrame(data)
+    binary_data = io.StringIO()
+    df.to_csv(binary_data, index=False, encoding="utf-8")
+    return binary_data.getvalue()
+
+
+def get_file_size(path: str) -> str:
+    return bytes_to_human_readable(Path(path).stat().st_size)
+
+
+def format_duration(seconds: float) -> str:
+    duration_units = [
+        ("week", 60 * 60 * 24 * 7),
+        ("day", 60 * 60 * 24),
+        ("hour", 60 * 60),
+        ("minute", 60),
+        ("second", 1),
+    ]
+
+    parts = []
+    for unit, duration in duration_units:
+        count, seconds = divmod(int(seconds), duration)
+        if count:
+            parts.append(f'{count} {unit}{"s" if count > 1 else ""}')
+
+    return ", ".join(parts) if parts else "0 seconds"

src/core/utils/translations.py

@@ -0,0 +1,4 @@
+translations = {
+    "en_US": {},
+    "zh_CN": {},
+}

src/features/admin/restapi.py → src/features/admin/api.py

@@ -13,8 +13,8 @@
 from src.deps import auth, get_session
 from src.features.admin import schemas
 from src.features.admin.models import Group, Permission, Role, User
-from src.features.admin.repository import group_repo, menu_repo, permission_repo, role_repo, user_repo
 from src.features.admin.security import generate_access_token_response
+from src.features.admin.services import group_repo, menu_repo, permission_repo, role_repo, user_repo
 
 router = APIRouter()