feat(backend): initialisation du projet FastAPI

Structure en couches api / services / repositories / models, sens de
dependance unique, une session SQLAlchemy async injectee par dependance.

- Python 3.14, dependances gerees par uv et verrouillees dans uv.lock
- FastAPI expose par une factory : aucune configuration lue a l'import,
  ce qui rend tests et migrations independants de l'environnement
- Settings Pydantic, APP_SECRET_KEY et DATABASE_URL sans valeur par defaut
- Sondes /health/live et /health/ready, metriques Prometheus sur /metrics
- Lint et format ruff, mypy strict, pytest avec couverture
- Alembic branche sur DATABASE_URL et non sur alembic.ini
- Image Docker multi-stage, utilisateur non root, sonde de sante integree
This commit is contained in:
Johan LEROY
2026-09-14 12:26:20 +02:00
parent 26704831e4
commit 6161a432c3
36 changed files with 1795 additions and 0 deletions
View File
View File
+10
View File
@@ -0,0 +1,10 @@
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import Settings, get_settings
from app.db.session import get_session
SessionDep = Annotated[AsyncSession, Depends(get_session)]
SettingsDep = Annotated[Settings, Depends(get_settings)]
View File
@@ -0,0 +1,33 @@
from fastapi import APIRouter, HTTPException, status
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
from app.api.deps import SessionDep, SettingsDep
from app.core.logging import get_logger
from app.schemas.health import LivenessStatus, ReadinessStatus
logger = get_logger(__name__)
router = APIRouter(tags=["health"])
@router.get("/live", summary="Sonde de vivacite")
async def liveness(settings: SettingsDep) -> LivenessStatus:
return LivenessStatus(
status="ok",
service=settings.name,
version=settings.version,
environment=settings.env,
)
@router.get("/ready", summary="Sonde de disponibilite")
async def readiness(session: SessionDep) -> ReadinessStatus:
try:
await session.execute(text("SELECT 1"))
except SQLAlchemyError, OSError:
logger.exception("Base de donnees injoignable")
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Base de donnees injoignable",
) from None
return ReadinessStatus(status="ready", database="reachable")
+6
View File
@@ -0,0 +1,6 @@
from fastapi import APIRouter
from app.api.v1.endpoints import health
api_router = APIRouter()
api_router.include_router(health.router, prefix="/health")
View File
+41
View File
@@ -0,0 +1,41 @@
from functools import lru_cache
from typing import Literal
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
Environment = Literal["local", "dev", "staging", "prod"]
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_prefix="APP_",
env_file_encoding="utf-8",
extra="ignore",
)
name: str = "EnerVision API"
version: str = "0.1.0"
env: Environment = "local"
debug: bool = False
log_level: str = "INFO"
api_prefix: str = "/api/v1"
secret_key: SecretStr
cors_origins: str = ""
database_url: str = Field(validation_alias="DATABASE_URL")
database_pool_size: int = 5
database_max_overflow: int = 10
@property
def allowed_origins(self) -> list[str]:
return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()]
@property
def is_production(self) -> bool:
return self.env == "prod"
@lru_cache
def get_settings() -> Settings:
return Settings()
+48
View File
@@ -0,0 +1,48 @@
import logging
from logging.config import dictConfig
from app.core.config import Settings
def configure_logging(settings: Settings) -> None:
formatter = "json" if settings.is_production else "console"
dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"console": {
"format": "%(asctime)s %(levelname)-8s %(name)s %(message)s",
},
"json": {
"()": "pythonjsonlogger.json.JsonFormatter",
"format": "%(asctime)s %(levelname)s %(name)s %(message)s",
},
},
"handlers": {
"default": {
"class": "logging.StreamHandler",
"formatter": formatter,
"stream": "ext://sys.stdout",
},
},
"root": {"handlers": ["default"], "level": settings.log_level},
"loggers": {
"uvicorn": {
"handlers": ["default"],
"level": settings.log_level,
"propagate": False,
},
"uvicorn.access": {
"handlers": ["default"],
"level": settings.log_level,
"propagate": False,
},
"sqlalchemy.engine": {"level": "WARNING"},
},
}
)
def get_logger(name: str) -> logging.Logger:
return logging.getLogger(name)
View File
+5
View File
@@ -0,0 +1,5 @@
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
"""Base declarative commune a tous les modeles."""
+33
View File
@@ -0,0 +1,33 @@
from collections.abc import AsyncIterator
from functools import lru_cache
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from app.core.config import get_settings
@lru_cache
def get_engine() -> AsyncEngine:
settings = get_settings()
return create_async_engine(
settings.database_url,
echo=settings.debug,
pool_pre_ping=True,
pool_size=settings.database_pool_size,
max_overflow=settings.database_max_overflow,
)
@lru_cache
def get_session_factory() -> async_sessionmaker[AsyncSession]:
return async_sessionmaker(get_engine(), class_=AsyncSession, expire_on_commit=False)
async def get_session() -> AsyncIterator[AsyncSession]:
async with get_session_factory()() as session:
yield session
+54
View File
@@ -0,0 +1,54 @@
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from prometheus_fastapi_instrumentator import Instrumentator
from app.api.v1.router import api_router
from app.core.config import Settings, get_settings
from app.core.logging import configure_logging, get_logger
from app.db.session import get_engine
logger = get_logger(__name__)
@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
settings = get_settings()
logger.info(
"Demarrage de %s %s en environnement %s", settings.name, settings.version, settings.env
)
yield
await get_engine().dispose()
def create_app(settings: Settings | None = None) -> FastAPI:
resolved = settings or get_settings()
configure_logging(resolved)
application = FastAPI(
title=resolved.name,
version=resolved.version,
debug=resolved.debug,
lifespan=lifespan,
docs_url=None if resolved.is_production else "/docs",
redoc_url=None if resolved.is_production else "/redoc",
openapi_url=None if resolved.is_production else "/openapi.json",
)
if resolved.allowed_origins:
application.add_middleware(
CORSMiddleware,
allow_origins=resolved.allowed_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Instrumentator().instrument(application).expose(
application, endpoint="/metrics", include_in_schema=False
)
application.include_router(api_router, prefix=resolved.api_prefix)
return application
+2
View File
@@ -0,0 +1,2 @@
# Piege : tout modele absent de ce module reste invisible de `alembic revision
# --autogenerate`, qui genererait alors un drop de sa table.
+3
View File
@@ -0,0 +1,3 @@
from app.schemas.health import LivenessStatus, ReadinessStatus
__all__ = ["LivenessStatus", "ReadinessStatus"]
+15
View File
@@ -0,0 +1,15 @@
from typing import Literal
from pydantic import BaseModel
class LivenessStatus(BaseModel):
status: Literal["ok"]
service: str
version: str
environment: str
class ReadinessStatus(BaseModel):
status: Literal["ready"]
database: Literal["reachable"]