Modèles FastAPI
Développement moderne et prêt pour la production avec FastAPI : disposition du projet, schémas Pydantic v2, injection de dépendances, motifs asynchrones, authentification, méthodes de service transactionnelles et tests.
Structure du projet
my_app/
|-- app/
| |-- main.py # Fabrique d'application, cycle de vie, middleware
| |-- config.py # Paramètres via pydantic-settings
| |-- dependencies.py # Dépendances partagées FastAPI
| |-- database.py # Moteur SQLAlchemy + session
| |-- routers/
| | `-- users.py
| |-- models/ # Modèles ORM SQLAlchemy
| | `-- user.py
| |-- schemas/ # Schémas de requête/réponse Pydantic
| | `-- user.py
| `-- services/ # Couche de logique métier
| `-- user_service.py
|-- tests/
| |-- conftest.py
| `-- test_users.py
|-- pyproject.toml
`-- .env
Fabrique d'application et cycle de vie
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.database import engine, Base
from app.routers import users
@asynccontextmanager
async def lifespan(app: FastAPI):
# Créer automatiquement les tables au démarrage pour faciliter l'utilisation dans les environnements de développement/démo.
# Pour les applications de production strictes, gérez les schémas via les migrations Alembic à la place.
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Arrêt : fermer les ressources mises en commun.
await engine.dispose()
def create_app() -> FastAPI:
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins,
allow_credentials=settings.allow_credentials,
allow_methods=settings.allowed_methods,
allow_headers=settings.allowed_headers,
)
app.include_router(users.router, prefix="/users", tags=["users"])
return app
app = create_app()
Configuration avec pydantic-settings
# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
app_name: str = "Mon App"
app_version: str = "0.1.0"
debug: bool = False
database_url: str
secret_key: str
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
# pydantic-settings v2 évalue directement les littéraux de liste mutables en toute sécurité
allowed_origins: list[str] = ["http://localhost:3000"]
allowed_methods: list[str] = ["GET", "POST", "PATCH", "DELETE", "OPTIONS"]
allowed_headers: list[str] = ["Authorization", "Content-Type"]
allow_credentials: bool = True
settings = Settings()
Schémas Pydantic (v2)
# app/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, Field, model_validator
class UserBase(BaseModel):
email: EmailStr
username: str = Field(min_length=3, max_length=50)
class UserCreate(UserBase):
password: str = Field(min_length=8)
password_confirm: str
@model_validator(mode="after")
def passwords_match(self) -> "UserCreate":
if self.password != self.password_confirm:
raise ValueError("Les mots de passe ne correspondent pas")
return self
class UserUpdate(BaseModel):
username: str | None = Field(default=None, min_length=3, max_length=50)
email: EmailStr | None = None
class UserResponse(UserBase):
id: int
is_active: bool
created_at: datetime
model_config = {"from_attributes": True}
class UserListResponse(BaseModel):
total: int
items: list[UserResponse]
Injection de dépendances
# app/dependencies.py
from typing import Annotated, AsyncGenerator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.database import AsyncSessionLocal
from app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/token")
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
except Exception:
await session.rollback()
raise
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Impossible de valider les identifiants",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
subject = payload.get("sub")
if subject is None:
raise credentials_exception
user_id = int(subject)
except (JWTError, TypeError, ValueError):
raise credentials_exception
user = await db.get(User, user_id)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user)],
) -> User:
if not current_user.is_active:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Utilisateur inactif")
return current_user
DbDep = Annotated[AsyncSession, Depends(get_db)]
CurrentUserDep = Annotated[User, Depends(get_current_user)]
ActiveUserDep = Annotated[User, Depends(get_current_active_user)]
Conception des routeurs et des points d'accès
# app/routers/users.py
from typing import Annotated
from fastapi import APIRouter, HTTPException, Query, status
from fastapi.security import OAuth2PasswordRequestForm
from app.dependencies import ActiveUserDep, DbDep
from app.schemas.user import UserCreate, UserResponse, UserUpdate, UserListResponse
from app.services.user_service import DuplicateUserError, UserService
router = APIRouter()
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, db: DbDep) -> UserResponse:
service = UserService(db)
try:
return await service.create(payload)
except DuplicateUserError:
raise HTTPException(status_code=400, detail="Email déjà enregistré")
@router.get("/me", response_model=UserResponse)
async def get_me(current_user: ActiveUserDep) -> UserResponse:
return current_user
@router.get("/", response_model=UserListResponse)
async def list_users(
db: DbDep,
current_user: ActiveUserDep,
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> UserListResponse:
service = UserService(db)
users, total = await service.list(skip=skip, limit=limit)
return UserListResponse(total=total, items=users)
@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
user_id: int,
payload: UserUpdate,
db: DbDep,
current_user: ActiveUserDep,
) -> UserResponse:
if current_user.id != user_id:
raise HTTPException(status_code=403, detail="Non autorisé")
service = UserService(db)
try:
user = await service.update(user_id, payload)
except DuplicateUserError:
raise HTTPException(status_code=400, detail="Email déjà enregistré")
if user is None:
raise HTTPException(status_code=404, detail="Utilisateur introuvable")
return user
@router.post("/token")
async def login(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
db: DbDep,
) -> dict[str, str]:
service = UserService(db)
token = await service.authenticate(form_data.username, form_data.password)
if token is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nom d'utilisateur ou mot de passe incorrect",
headers={"WWW-Authenticate": "Bearer"},
)
return {"access_token": token, "token_type": "bearer"}
Couche de service
# app/services/user_service.py
from datetime import datetime, timedelta, timezone
from jose import jwt
from passlib.context import CryptContext
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class DuplicateUserError(Exception):
"""Levée lorsqu'un champ utilisateur unique entre en conflit avec une ligne existante."""
class UserService:
def __init__(self, db: AsyncSession) -> None:
self.db = db
async def get_by_email(self, email: str) -> User | None:
result = await self.db.execute(select(User).where(User.email == email))
return result.scalar_one_or_none()
async def create(self, payload: UserCreate) -> User:
user = User(
email=payload.email,
username=payload.username,
hashed_password=pwd_context.hash(payload.password),
)
self.db.add(user)
try:
# S'appuyer sur les contraintes atomiques de la base de données plutôt que sur des pré-vérifications au niveau de l'application sujettes aux races conditions
await self.db.commit()
except IntegrityError as exc:
await self.db.rollback()
raise DuplicateUserError from exc
await self.db.refresh(user)
return user
async def list(self, skip: int = 0, limit: int = 20) -> tuple[list[User], int]:
total_result = await self.db.execute(select(func.count(User.id)))
total = total_result.scalar_one()
# Appliquer un ordre déterministe explicite pour garantir une pagination fiable
result = await self.db.execute(
select(User).order_by(User.id).offset(skip).limit(limit)
)
return list(result.scalars()), total
async def update(self, user_id: int, payload: UserUpdate) -> User | None:
user = await self.db.get(User, user_id)
if user is None:
return None
for field, value in payload.model_dump(exclude_unset=True).items():
setattr(user, field, value)
try:
await self.db.commit()
except IntegrityError as exc:
await self.db.rollback()
raise DuplicateUserError from exc
await self.db.refresh(user)
return user
async def authenticate(self, email: str, password: str) -> str | None:
user = await self.get_by_email(email)
if user is None or not pwd_context.verify(password, user.hashed_password):
return None
expire = datetime.now(timezone.utc) + timedelta(
minutes=settings.access_token_expire_minutes
)
return jwt.encode(
{"sub": str(user.id), "exp": expire},
settings.secret_key,
algorithm=settings.algorithm,
)
Note sur la conception de la base de données : La gestion de l'unicité au niveau de l'application nécessite un index unique sous-jacent dans la base de données (par exemple,
unique=Truesur vos attributs de mappage SQLAlchemy). Sans contraintes sous-jacentes, la capture d'erreur au niveau de l'application ne peut pas empêcher en toute sécurité les conditions de concurrence.
Tests avec httpx et pytest
# tests/conftest.py
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.database import Base
from app.dependencies import get_db
from app.main import create_app
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
engine = create_async_engine(TEST_DATABASE_URL)
TestingSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
@pytest_asyncio.fixture(autouse=True)
async def setup_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def db_session():
async with TestingSessionLocal() as session:
yield session
await session.rollback()
@pytest_asyncio.fixture
async def client(db_session: AsyncSession):
app = create_app()
async def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
yield ac
@pytest_asyncio.fixture
async def registered_user(client: AsyncClient) -> dict:
resp = await client.post("/users/", json={
"email": "[email protected]",
"username": "testuser",
"password": "securepass1",
"password_confirm": "securepass1",
})
assert resp.status_code == 201
return resp.json()
@pytest_asyncio.fixture
async def auth_token(client: AsyncClient, registered_user: dict) -> str:
resp = await client.post("/users/token", data={
"username": "[email protected]",
"password": "securepass1",
})
assert resp.status_code == 200
return resp.json()["access_token"]
@pytest_asyncio.fixture
async def auth_client(client: AsyncClient, auth_token: str) -> AsyncClient:
client.headers.update({"Authorization": f"Bearer {auth_token}"})
return client
Anti-patrons
# Mauvais : logique métier à l'intérieur des gestionnaires de route.
@router.post("/users/")
async def create_user(payload: UserCreate, db: DbDep):
hashed = bcrypt.hash(payload.password)
user = User(email=payload.email, hashed_password=hashed)
db.add(user)
await db.commit()
return user
# Bon : route légère, gestion transactionnelle du service.
@router.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(payload: UserCreate, db: DbDep):
try:
return await UserServ
…[tronqué pour la traduction]