API FastAPI avancée : sécurisée et scalable 2026

Introduction

En 2026, FastAPI reste le framework Python leader pour les APIs performantes grâce à son support natif de l'asynchrone, sa validation automatique via Pydantic et sa génération OpenAPI interactive. Ce tutoriel avancé vous guide dans la construction d'une API complète pour gérer des utilisateurs : CRUD sécurisé, authentification JWT, base de données asynchrone avec SQLAlchemy, middlewares personnalisés pour le rate limiting et CORS, ainsi que des tests exhaustifs avec pytest.

Pourquoi c'est crucial ? Les APIs modernes doivent scaler horizontalement, résister aux attaques (DDoS, injections) et intégrer seamlessly avec des frontends React/Vue ou des microservices. Contrairement aux tutoriels basiques, nous plongeons dans les patterns production-ready : dépendances hiérarchiques, Alembic pour migrations, et monitoring avec logging structuré. À la fin, vous aurez un projet copier-collable, déployable sur Docker, prêt pour AWS ou Vercel. Comptez 30 minutes pour le setup et les tests – un gain de temps énorme pour un dev senior.

Prérequis

Python 3.12+
Connaissances avancées en Python async/await et typing
Familiarité avec SQLAlchemy, Pydantic et JWT
Outils : Poetry (gestionnaire de dépendances), Alembic (migrations), pytest (tests)
IDE avec support LSP (VS Code + Ruff)

Installation du projet

terminal

mkdir fastapi-advanced-api && cd fastapi-advanced-api
poetry init --no-interaction --python "^3.12"
poetry add fastapi uvicorn[standard] sqlalchemy[asyncio] aiosqlite pydantic[email] python-jose[cryptography] passlib[bcrypt] python-multipart httpx
poetry add --group dev pytest pytest-asyncio pytest-httpx alembic ruff black
poetry install
poetry shell
alembic init alembic

Ce script initialise un projet Poetry pour une isolation parfaite des dépendances. On installe FastAPI core, une DB async légère (aiosqlite), l'auth (JWT + bcrypt), et les dev tools pour tests/migrations. Alembic prépare les migrations DB – évitez pip pour prévenir les conflits de versions en prod.

Configuration de la base de données async

app/database.py

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from typing import AsyncGenerator

DATABASE_URL = "sqlite+aiosqlite:///./app.db"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Ce module crée un moteur async SQLAlchemy avec aiosqlite pour des queries non-bloquantes. La dépendance get_db() injecte une session par requête, avec rollback auto sur erreur – pattern essentiel pour éviter les fuites de connexions en high-load. echo=True aide au debug, désactivez en prod.

Modèles SQLAlchemy et schémas Pydantic

app/models.py

from sqlalchemy import String, Integer, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String, nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    def __repr__(self) -> str:
        return f"User(id={self.id!r}, email={self.email!r})"

Modèle User avec typing moderne (Mapped). Les index sur email/id boostent les queries. Pas de schéma Pydantic ici – ils sont séparés pour découpler DB et API, évitant les fuites de données sensibles comme hashed_password dans les réponses.

Schémas Pydantic pour validation

app/schemas.py

from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime


class UserBase(BaseModel):
    email: EmailStr


class UserCreate(UserBase):
    password: str = Field(min_length=8)


class User(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    email: Optional[str] = None

Schémas hiérarchiques pour inputs/outputs. EmailStr valide les emails, Field enforce les règles. from_attributes=True mappe ORM vers Pydantic sans boilerplate. Séparez toujours Create/Read pour masquer hashed_password et prévenir les OWASP Top10 breaches.

Logique CRUD et auth JWT

app/crud.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from passlib.context import CryptContext
from jose import JWTError, jwt

from app.models import User
from app.schemas import UserCreate, User

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = "your-secret-key-change-in-prod"
ALGORITHM = "HS256"


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()


async def create_user(db: AsyncSession, user: UserCreate) -> User:
    hashed_password = pwd_context.hash(user.password)
    db_user = User(email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    await db.flush()
    return db_user


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def create_access_token(data: dict):
    to_encode = data.copy()
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def authenticate_user(db: AsyncSession, email: str, password: str):
    user = await get_user_by_email(db, email)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

CRUD async minimal mais robuste : scalar_one_or_none() gère l'unicité sans exceptions inutiles. Bcrypt pour hash sécurisé, JWT HS256 pour tokens (changez SECRET_KEY !). authenticate_user centralise la vérif – réutilisable dans guards OAuth2.

Routeur users avec dépendances

app/routers/users.py

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

from app.database import get_db
from app.crud import authenticate_user, create_user, create_access_token
from app.schemas import User, UserCreate, Token

router = APIRouter(prefix="/users", tags=["users"])
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="users/token")


@router.post("/", response_model=User)
async def create_new_user(user: UserCreate, db=Depends(get_db)):
    db_user = await get_user_by_email(db, user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return await create_user(db, user)


@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db=Depends(get_db)):
    user = await authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect credentials")
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

Routeur modulaire pour scalabilité (un par entité). OAuth2PasswordBearer pour Bearer tokens, Depends(get_db) injecte la session. Guards implicites via exceptions HTTP – génère auto Swagger UI. Ajoutez @router.get("/me") avec current_user: User = Depends(get_current_user) pour protected routes.

Middlewares et app principale

app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security.utils import get_authorization_scheme_param

from app.database import engine, Base, get_db
from app.routers import users


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict in prod
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    # Simple in-memory limit: 10 req/min (use Redis in prod)
    if not hasattr(app.state, 'requests'):
        app.state.requests = {}
    now = request.url.path
    app.state.requests[client_ip] = app.state.requests.get(client_ip, []) + [now]
    recent = [r for r in app.state.requests[client_ip] if now - r < 60]
    if len(recent) > 10:
        raise HTTPException(429, "Rate limit exceeded")
    app.state.requests[client_ip] = recent
    response = await call_next(request)
    return response

app.include_router(users.router)

@app.get("/health")
async def health():
    return {"status": "healthy"}

App lifecycle avec lifespan pour init DB auto. Middleware CORS et rate-limit custom (in-memory, migrez vers Redis). include_router pour modularité. Lancez avec uvicorn app.main:app --reload – docs auto à /docs.

Tests unitaires avec pytest

tests/test_users.py

import pytest
from httpx import AsyncClient
from app.main import app


@pytest.fixture(scope="module")
async def client():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users/", json={"email": "test@example.com", "password": "testpass123"})
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "test@example.com"


@pytest.mark.asyncio
async def test_login(client: AsyncClient):
    response = await client.post("/users/token", data={"username": "test@example.com", "password": "testpass123"})
    assert response.status_code == 200
    assert "access_token" in response.json()

Tests async avec pytest-asyncio et httpx.AsyncClient pour simuler requêtes réelles. Fixture client réutilisable. Couvre happy paths – ajoutez mocks pour DB en CI/CD. Couverture >90% avec pytest --cov.

Bonnes pratiques

Migrations avec Alembic : Configurez alembic.ini et env.py pour versioning DB. alembic revision --autogenerate -m "add users".
Environ vars : Utilisez python-dotenv pour SECRET_KEY, DB_URL.
Logging structuré : Intégrez structlog pour traces JSON en prod.
Validation custom : Pydantic validators pour business rules (ex: password strength).
Caching : Redis + fastapi-cache pour endpoints read-heavy.

Erreurs courantes à éviter

Oublier await sur DB ops : provoque des RuntimeWarning et data loss.
Exposer hashed_password dans schémas response : toujours filtrer via response_model.
Rate limit in-memory : non-scalable, passez à Redis ou middleware tiers comme slowapi.
Pas de token expiration : Ajoutez exp claim JWT (15min) + refresh tokens.

Pour aller plus loin

Approfondissez avec nos formations Learni sur FastAPI avancé. Ressources : Docs FastAPI, SQLAlchemy Async, Typer pour CLI. Déployez sur Render/K8s avec Dockerfile fourni.

Comment implémenter une API FastAPI avancée en 2026