Après plusieurs mois d'utilisation intensive de FastAPI en environnement de production, j'ai développé une structure de projet standardisée qui optimise à la fois la maintenabilité et l'extensibilité. Cette architecture permet de gérer efficacement les évolutions du code tout en maintenant des standards de sécurité élevés. Dans ce guide, je détaille chaque composant de cette architecture "référence" que j'utilise systématiquement lors du démarrage de nouveaux projets ou lors de refactoring d'applications existantes.
Structure du Projet
Répertoire App
Le répertoire app constitue le cœur de l'application et contient l'ensemble des modules nécessaires à l'ifnrastructure backend. Voici le détail de chaque sous-répertoire et leur utilité.
__init__.py — Ce fichier est indispensable dans chaque package Python. Il permet de déclarer le répertoire comme un package Python valide et d'activer les imports relatifs.
api_router.py — Ce module centralise la gestion de tous les routeurs de l'application. Cette approche facilite considérablement le contrôle de版本 (comme /v1/utilisateurs, /v2/utilisateurs) et maintient une organisation claire du code.
# apps/api_router.py
from fastapi import APIRouter
api_v1 = APIRouter(prefix="/v1")
api_v2 = APIRouter(prefix="/v2")
# Intégration des routeurs
# from app.routers import routers as user_routers
# api_v1.include_router(users_routers)
logger.py — Le logging est un élément crucial pour tout logiciel. Il offre une visibilité complète sur l'exécution et facilite considérablement le débogage. Cette configuration utilise TimedRotatingFileHandler pour une rotation quotidienne des journaux.
# apps/logger.py
import logging
import json
from logging.handlers import TimedRotatingFileHandler
logger = logging.getLogger()
class JsonFormatter(logging.Formatter):
def format(self, record):
log_record = {
"timestamp": self.formatTime(record, self.datefmt),
"level": record.levelname,
"module": record.module,
"funcName": record.funcName,
"lineno": record.lineno,
"message": record.getMessage(),
}
return json.dumps(log_record)
file_handler = TimedRotatingFileHandler(
"logs/app.log", when="midnight", interval=1/86400, backupCount=7
)
file_handler.setFormatter(JsonFormatter())
logger.handlers = [file_handler]
logger.setLevel(logging.INFO)
main.py — Le point d'entrée principal de l'application où sont définies les politiques de sécurité et les règles globales.
# apps/main.py
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from fastapi import FastAPI
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse, JSONResponse
from fastapi.requests import Request
from fastapi import HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address
from redis import asyncio as aioredis
from app.logger import logger
from app.api_router import api
from app.settings import Settings
from app.middlewares import log_request_middleware
settings = Settings()
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
def initiate_app():
app = FastAPI(
title="API Projet Modulaire",
summary="API pour projet FastAPI modulaire",
lifespan=lifespan,
)
origins = []
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=[],
)
app.add_middleware(BaseHTTPMiddleware, dispatch=log_request_middleware)
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.include_router(api)
return app
app = initiate_app()
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={
"detail": exc.detail,
"path": request.url.path,
"timestamp": datetime.now(timezone.utc).isoformat(),
},
)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Erreur inattendue: {str(exc)}")
return JSONResponse(
status_code=500,
content={
"detail": "Une erreur inattendue s'est produite",
"path": request.url.path,
},
)
@app.get("/", tags=["Racine"])
async def root():
return RedirectResponse("/docs")
settings.py — L'utilisation de Pydantic BaseSettings pour la gestion des variables d'environnement est à la fois simple et robuste. Les types définis sont validés au démarrage de l'application.
# apps/settings.py
from pydantic_settings import BaseSettings
from pydantic import ConfigDict, AnyUrl
class Settings(BaseSettings):
model_config = ConfigDict(env_file=".env")
api_secret: str = "valeur_par_defaut"
JWT_SECRET: str
JWT_ALGORITHM: str = "HS256"
DATABASE_URL: AnyUrl
dependencies.py — centralise les dépendances personnalisées des routeurs. Les dépendances communes comme get_db peuvent être déclarées ici, améliorant ainsi l'organisation et la testabilité du code.
middlewares.py — Ce module gère les middlewares personnalisés de l'application. Les bénéfices en termes de modularité sont identiques à ceux des dépendances.
Répertoire Routers
Je recommande de créer des modules distincts pour chaque groupe logique de routes. Par exemple, un fichier auth.py pour les routes d'authentification, products.py pour la gestion des produits, et admin.py pour les fonctionnalités administratives. Cette approche maintient les fonctions de route concises (2 lignes maximum), deleguant la logique métier aux services.
# apps/routers/auth.py
from typing import Annotated
from pydantic import EmailStr
from fastapi.routing import APIRouter
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends, Body, BackgroundTasks
from app.dependencies import get_db
from app.schemas import auth as auth_schemas
from app.services import auth as auth_services
router = APIRouter(prefix="/auth", tags=["Authentification"])
EmailBody = Annotated[EmailStr, Body(embed=True)]
DBDep = Annotated[AsyncSession, Depends(get_db)]
@router.post("/inscription", response_model=auth_schemas.UserModel)
async def inscription(
db: DBDep,
bg_task: BackgroundTasks,
request_data: auth_schemas.UserSignUpData,
):
return await auth_services.create_user_account(request_data, db, bg_task)
Répertoire Schemas
Tous les modèles Pydantic sont regroupés dans le module schemas. En maintenant une correspondance entre les modules de routes, les schemas et les services, l'application gagne en cohérence et en prédictibilité.
# apps/schemas/auth.py
from uuid import UUID
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, EmailStr, Field
class UserSignUpData(BaseModel):
password: Annotated[str, Field(min_length=8)]
email: Annotated[EmailStr, Field(max_length=254)]
class UserModel(BaseModel):
id: UUID
email: EmailStr
date_creation: datetime
date_mise_a_jour: datetime
Répertoire Services
Ce répertoire contient la logique métier réelle : appels aux APIs externes, requêtes数据库, etc. Cette séparation claire entre la déclaration des routes (aspect déclaratif) et leur implémentation (aspect fonctionnel) est fonadmentale. Voici un exemple avec le composant auth :
# apps/services/auth.py
from datetime import timedelta
from fastapi import HTTPException
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.settings import Settings
from app.models import User as UserDB
from app.schemas import auth as auth_schema
settings = Settings()
SECRET_KEY = settings.JWT_SECRET
ALGORITHME = settings.JWT_ALGORITHM
TOKEN_EXPIRATION = timedelta(heures=48)
REFRESH_EXPIRATION = timedelta(jours=5)
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="v1/auth/token")
def verify_password(plain_pwd: str, hashed_pwd: str):
return pwd_context.verify(plain_pwd, hashed_pwd)
def hash_password(password: str):
return pwd_context.hash(password)
async def retrieve_user(email: str, session: AsyncSession) -> UserDB | None:
stmt = select(UserDB).where(UserDB.email == email)
result = await session.execute(stmt)
return result.scalar_one_or_none()
async def create_user_account(
user_data: auth_schema.UserSignUpData,
session: AsyncSession,
):
verification = await session.execute(
select(UserDB).where(UserDB.email == user_data.email)
)
if verification.scalar_one_or_none():
raise HTTPException(status_code=400, detail="Email déjà enregistré")
hashed_password = hash_password(user_data.password)
nouveau_utilisateur = UserDB(
email=user_data.email,
password=hashed_password,
username=user_data.email,
)
session.add(nouveau_utilisateur)
await session.commit()
await session.refresh(nouveau_utilisateur)
return nouveau_utilisateur
Automatisation avec Makefile
L'utilisation d'un Makefile simplifie considérablement l'exécution des commandes récurrentes comme le démarrage du serveur ou le lancement des tests.
# Makefile
run-local:
fastapi dev app/main.py
test-local:
pytest -s --cov
coverage-report:
coverage report
coverage-html:
coverage report && coverage html
Avec ce fichier à la racine du projet, il suffit d'exécuter make run-local pour lancer le serveur de développement.