Pydantic v2 (validation, models, settings)

TL;DR — Pydantic v2 est la couche de validation runtime de l'écosystème Python moderne : tu déclares la forme de tes données avec des annotations de type, et Pydantic les valide, coerce et sérialise à la frontière de ton système (HTTP, queue, LLM, base de données). Pense à class-validator + class-transformer de NestJS, mais fusionnés dans un seul objet, avec un cœur compilé en Rust (pydantic-core) qui le rend 5 à 50× plus rapide que la v1. En v2 tu écris model_validate, model_dump, Field, des @field_validator/@model_validator, et tu utilises Annotated pour composer des contraintes réutilisables. Pour construire des agents IA, c'est l'outil qui transforme une réponse LLM en un objet typé fiable : tu passes ton modèle Pydantic à client.messages.parse() de l'Anthropic SDK et tu récupères un objet validé, pas du JSON à try/except.

🧠 Mental model

Tu connais déjà cette idée, juste sous d'autres noms. En NestJS, un DTO décoré avec class-validator filtre les requêtes au niveau du ValidationPipe. En TypeScript, Zod fait la validation runtime que tsc ne fait pas. Pydantic est exactement ce rôle en Python — mais c'est devenu le standard de fait : FastAPI, l'Anthropic SDK, LangChain, OpenAI SDK, SQLModel, et la moitié de l'écosystème data en dépendent.

Le piège mental venant de TypeScript : en TS, les types disparaissent à la compilation. interface User { age: number } ne vérifie rien au runtime — si une API te renvoie age: "42", ton code plante plus loin, mystérieusement. Les annotations Python sont pareilles : par défaut, age: int ne valide rien. Pydantic est la machine qui fait respecter ces annotations au runtime.

L'analogie clé : Pydantic est un poste-frontière, pas un système de types interne.

   Monde extérieur (non fiable)          Ton domaine (fiable, typé)
  ┌──────────────────────────┐         ┌─────────────────────────┐
  │  JSON HTTP                │         │                         │
  │  payload de queue         │  ─────▶ │   User(age=42)          │
  │  réponse LLM              │ valide  │   age: int (garanti)    │
  │  variables d'env          │ coerce  │   email: str (validé)   │
  │  ligne de DB              │ rejette │                         │
  └──────────────────────────┘         └─────────────────────────┘
        str, dict, bytes...            objets Python avec invariants

Tu valides une fois, à l'entrée. Après, à l'intérieur de ton code, tu fais confiance aux types. Tu ne refais pas if isinstance(...) partout — c'est le boulot de la frontière. C'est exactement le « parse, don't validate » que tu as peut-être croisé en fonctionnel.

Le modèle de base

Un BaseModel déclare des champs avec des annotations. À l'instanciation, Pydantic valide.

from pydantic import BaseModel, EmailStr, Field


class User(BaseModel):
    id: int
    email: EmailStr
    full_name: str = Field(min_length=1, max_length=120)
    is_active: bool = True
    tags: list[str] = []


# Validation + coercion à la construction
user = User(id="42", email="[email protected]", full_name="Alice")
# id="42" (str) devient 42 (int) — coercion contrôlée
assert user.id == 42 and isinstance(user.id, int)

# Construire depuis un dict non fiable (ce que tu fais le plus souvent)
raw: dict = {"id": 7, "email": "[email protected]", "full_name": "Bob"}
user2 = User.model_validate(raw)

Et la sérialisation, le chemin retour :

user.model_dump()
# {'id': 42, 'email': '[email protected]', 'full_name': 'Alice',
#  'is_active': True, 'tags': []}

user.model_dump_json()           # -> str JSON
User.model_json_schema()         # -> JSON Schema (utile pour les LLM, voir plus bas)

⚠️ Piège de migration v1 → v2. Toute la nomenclature a changé. Si tu trouves de vieux exemples, traduis :

Pydantic v1	Pydantic v2
.dict()	.model_dump()
.json()	.model_dump_json()
.parse_obj(d)	.model_validate(d)
.parse_raw(s)	.model_validate_json(s)
@validator	@field_validator
@root_validator	@model_validator
class Config:	model_config = ConfigDict(...)
.schema()	.model_json_schema()

La bonne et la mauvaise façon de gérer une erreur

from pydantic import BaseModel, ValidationError


class Order(BaseModel):
    sku: str
    quantity: int = Field(gt=0)


# ❌ La mauvaise façon — valider à la main, façon PHP/JS d'il y a 10 ans
def parse_order_wrong(raw: dict) -> dict:
    if "sku" not in raw or not isinstance(raw["sku"], str):
        raise ValueError("sku invalide")
    if not isinstance(raw.get("quantity"), int) or raw["quantity"] <= 0:
        raise ValueError("quantity invalide")
    return raw  # toujours un dict, aucun type garanti en aval


# ✅ La façon idiomatique — déclaratif, et tu récupères un objet typé
def parse_order(raw: dict) -> Order:
    try:
        return Order.model_validate(raw)
    except ValidationError as exc:
        # exc.errors() est une liste structurée, parfaite pour une réponse 422
        raise BadRequest(detail=exc.errors()) from exc

ValidationError.errors() renvoie une liste de dicts (loc, msg, type, input) — c'est exactement le format que FastAPI renvoie en 422. Ne le re-formatte pas à la main.

Annotated : la composition de contraintes (idiome v2)

Le pattern le plus puissant et le plus sous-utilisé en v2 est Annotated. Il te permet de fabriquer des types réutilisables porteurs de leurs contraintes — l'équivalent d'un brand type Zod ou d'un type custom décoré.

from typing import Annotated
from pydantic import BaseModel, Field, StringConstraints

# Un type réutilisable, contraint, partageable dans tout le codebase
Slug = Annotated[str, StringConstraints(pattern=r"^[a-z0-9-]+$", min_length=1)]
PositiveInt = Annotated[int, Field(gt=0)]
Percentage = Annotated[float, Field(ge=0.0, le=1.0)]


class Product(BaseModel):
    slug: Slug
    stock: PositiveInt
    discount: Percentage = 0.0

Avantage senior : tu définis Slug une fois, tu le réutilises dans 20 modèles. Si la règle change, tu modifies un endroit. C'est de la DRY au niveau des types.

Validators : la logique métier à la frontière

Trois moments d'intervention. Le bon réflexe est de privilégier les contraintes déclaratives (Field, Annotated) et de réserver les validators à la logique qui ne s'exprime pas autrement.

from pydantic import BaseModel, field_validator, model_validator
from datetime import date


class Booking(BaseModel):
    check_in: date
    check_out: date
    guest_email: str

    # field_validator : transforme/valide UN champ
    @field_validator("guest_email")
    @classmethod
    def normalize_email(cls, v: str) -> str:
        return v.strip().lower()

    # model_validator(mode="after") : règle inter-champs, sur l'objet déjà validé
    @model_validator(mode="after")
    def check_dates(self) -> "Booking":
        if self.check_out <= self.check_in:
            raise ValueError("check_out doit être après check_in")
        return self

Distinction à retenir :

• mode="before" : reçoit la valeur brute (avant coercion). Utile pour nettoyer un format exotique ("1 000,50" → 1000.50).
• mode="after" (défaut) : reçoit la valeur déjà typée. C'est là que va 90 % de ta logique.
• Sur un model_validator(mode="after"), tu reçois self (instance complète) — idéal pour les invariants entre champs.

Pydantic dans FastAPI : la DI et les modèles

C'est ici que ton stack converge. FastAPI utilise Pydantic pour valider le body, sérialiser la réponse, et générer l'OpenAPI — automatiquement. Tu n'écris jamais de ValidationPipe, c'est intégré.

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field, EmailStr

app = FastAPI()


class CreateUserRequest(BaseModel):
    email: EmailStr
    full_name: str = Field(min_length=1, max_length=120)


class UserResponse(BaseModel):
    id: int
    email: EmailStr
    full_name: str


# Modèle d'entrée ≠ modèle de sortie. Sépare-les TOUJOURS.
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(payload: CreateUserRequest) -> UserResponse:
    # payload est déjà validé ici — un body invalide -> 422 automatique,
    # l'endpoint n'est jamais appelé.
    user_id = await insert_user(payload.email, payload.full_name)
    return UserResponse(id=user_id, email=payload.email, full_name=payload.full_name)

⚠️ Ne réponds jamais avec ton modèle d'ORM/de DB directement. Le response_model agit comme un filtre de sortie : il garantit que password_hash, internal_notes ou tout champ sensible n'est jamais sérialisé, même s'il existe sur l'objet renvoyé. Un modèle de requête et un modèle de réponse distincts, c'est une frontière de sécurité, pas de la cérémonie.

Settings avec pydantic-settings

La config typée, chargée depuis l'environnement, validée au démarrage. Le démarrage fail-fast sur une config invalide vaut bien mieux qu'un KeyError à 3h du matin.

from functools import lru_cache
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_")

    anthropic_api_key: str = Field(min_length=1)   # APP_ANTHROPIC_API_KEY
    model: str = "claude-opus-4-8"
    max_concurrency: int = Field(default=8, gt=0)
    request_timeout_s: float = 30.0


@lru_cache
def get_settings() -> Settings:
    return Settings()  # type: ignore[call-arg]  # rempli depuis l'env


# Dans FastAPI, c'est une dépendance injectable
from fastapi import Depends


@app.get("/health")
async def health(settings: Settings = Depends(get_settings)) -> dict[str, str]:
    return {"model": settings.model}

⚙️ En production : servir et appeler des agents LLM

C'est le cas d'usage qui te concerne directement. Un LLM renvoie du texte. Tu veux des objets typés. Pydantic + l'Anthropic SDK font le pont. Voici les patterns réels.

1. Structured outputs — messages.parse() avec ton modèle Pydantic

L'Anthropic SDK supporte les structured outputs natifs : tu passes un modèle Pydantic, le SDK contraint la sortie au JSON Schema et te renvoie un objet déjà validé. C'est l'inverse exact du json.loads() + try/except fragile.

from anthropic import AsyncAnthropic
from pydantic import BaseModel, Field

client = AsyncAnthropic()  # lit ANTHROPIC_API_KEY depuis l'env


class TicketTriage(BaseModel):
    category: str = Field(description="Une de : billing, technical, account, other")
    severity: int = Field(ge=1, le=5)
    summary: str = Field(max_length=200)
    needs_human: bool


async def triage(ticket_text: str) -> TicketTriage:
    response = await client.messages.parse(
        model="claude-opus-4-8",
        max_tokens=1024,
        output_format=TicketTriage,  # Pydantic -> JSON Schema, géré par le SDK
        messages=[{"role": "user", "content": ticket_text}],
    )
    # parsed_output est une instance TicketTriage validée, pas du JSON brut
    result = response.parsed_output
    if result is None:
        raise RuntimeError("le modèle a refusé ou n'a pas respecté le schéma")
    return result

# ❌ La mauvaise façon — encore vue partout, fragile
import json

async def triage_wrong(ticket_text: str) -> dict:
    response = await client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        messages=[{"role": "user", "content": f"Réponds en JSON: {ticket_text}"}],
    )
    text = response.content[0].text
    return json.loads(text)  # plante sur un préambule, un ``` ```, un champ manquant...

Le JSON Schema généré par model_json_schema() est exactement ce que les structured outputs consomment. Les contraintes (ge, le, max_length) deviennent des contraintes de schéma. Note : les structured outputs natifs ne supportent pas toutes les contraintes JSON Schema (pas de minimum/maximum numériques, pas de minLength/maxLength, pas de schémas récursifs) — le SDK Python retire silencieusement les contraintes non supportées du schéma envoyé et les revalide côté client. Donc tes Field(ge=1, le=5) sont quand même appliqués, mais par Pydantic au retour, pas forcément par le modèle.

2. Définir des tools d'agent à partir de modèles Pydantic

Une boucle tool-use a besoin d'un input_schema JSON Schema par tool. Tu le génères depuis Pydantic — une seule source de vérité pour le schéma et la validation des arguments que le modèle renvoie.

from anthropic import AsyncAnthropic
from anthropic.types import MessageParam
from pydantic import BaseModel, Field, ValidationError

client = AsyncAnthropic()


class GetWeatherArgs(BaseModel):
    city: str = Field(description="Ville et pays, ex: Paris, FR")
    unit: str = Field(default="celsius", description="celsius ou fahrenheit")


WEATHER_TOOL = {
    "name": "get_weather",
    "description": "Renvoie la météo actuelle d'une ville.",
    "input_schema": GetWeatherArgs.model_json_schema(),  # source unique
}


async def run_agent(user_msg: str) -> str:
    messages: list[MessageParam] = [{"role": "user", "content": user_msg}]

    while True:
        response = await client.messages.create(
            model="claude-opus-4-8",
            max_tokens=2048,
            thinking={"type": "adaptive"},  # adaptive thinking, jamais budget_tokens sur opus-4-8
            tools=[WEATHER_TOOL],
            messages=messages,
        )

        if response.stop_reason == "refusal":
            return "[refus de sécurité]"
        if response.stop_reason != "tool_use":
            # end_turn : on extrait le texte final
            return "".join(b.text for b in response.content if b.type == "text")

        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            try:
                # Le modèle a produit un dict — on le valide AVANT de l'exécuter.
                args = GetWeatherArgs.model_validate(block.input)
            except ValidationError as exc:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "is_error": True,
                    "content": f"Arguments invalides: {exc.errors()}",
                })
                continue

            weather = await fetch_weather(args.city, args.unit)  # ton implémentation
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": weather,
            })

        messages.append({"role": "user", "content": tool_results})

Le point senior : valide block.input avec Pydantic avant d'exécuter le tool. Le modèle peut halluciner un argument hors énum ou d'un mauvais type ; renvoyer is_error: True avec les errors() permet au modèle de se corriger au tour suivant — boucle de réparation gratuite.

3. Streaming + agrégation typée

Pour les longues sorties, stream pour éviter les timeouts HTTP. Tu accumules les tokens, puis tu valides l'objet complet à la fin.

async def triage_streamed(ticket_text: str) -> TicketTriage:
    async with client.messages.stream(
        model="claude-opus-4-8",
        max_tokens=1024,
        output_format=TicketTriage,
        messages=[{"role": "user", "content": ticket_text}],
    ) as stream:
        async for event in stream:
            ...  # pousse les deltas vers ton UI (SSE/WebSocket) si besoin
        final = await stream.get_final_message()
    # même garantie : objet validé à l'arrivée
    result = final.parsed_output
    if result is None:
        raise RuntimeError("le modèle a refusé ou n'a pas respecté le schéma")
    return result

Failure modes, perf et tradeoffs

• model_validate n'est pas gratuit. Le cœur Rust est rapide, mais dans une boucle chaude (des dizaines de milliers de validations/s) ça compte. Mesure. Pour valider une liste d'items homogènes, un TypeAdapter(list[Item]).validate_python(rows) est nettement plus rapide que valider item par item dans une boucle Python.
• Coercion silencieuse = bug subtil. Par défaut, id: int accepte "42" et 42.0. Si tu veux du strict (refuser "42"), utilise model_config = ConfigDict(strict=True) ou Annotated[int, Field(strict=True)]. Sur des données financières ou des frontières critiques, le mode strict évite des classes entières de bugs.
• model_dump(mode="json") vs model_dump(). Le premier produit des types JSON-sérialisables (un datetime devient une str ISO, un Decimal une str/float). Le second garde les objets Python. Si tu envoies dans une queue ou une réponse HTTP, utilise mode="json" ou directement model_dump_json().
• Sécurité — Field(exclude=True) et modèles de sortie. Ne sérialise jamais un secret. Modèle de réponse distinct du modèle interne (cf. FastAPI plus haut). Un champ password_hash sur ton modèle DB ne doit pas exister sur ton UserResponse.
• Observabilité. ValidationError.errors() est structuré (loc, type) — logue le type (ex: int_parsing, missing) en métrique, pas le message brut. Tu sauras quel champ casse en prod sans parser des strings. Pour les agents : logue le taux de tool_use rejetés par validation — c'est un signal direct de qualité de tes descriptions de tools.
• Coût LLM. Les structured outputs ajoutent une compilation de schéma (latence one-shot sur le premier appel d'un schéma donné, puis cache 24h côté Anthropic). Garde tes modèles stables pour profiter du cache. Combine avec le prompt caching (cache_control) sur le system prompt si tu sers beaucoup de requêtes au même agent.
• Tradeoff extra. ConfigDict(extra="forbid") rejette les champs inconnus (strict, bon pour les API internes et détecter les drifts de contrat). extra="ignore" (défaut) les laisse passer silencieusement. extra="allow" les conserve. Sur une réponse LLM, forbid est souvent trop agressif (le modèle peut ajouter un champ) ; sur un body HTTP interne, forbid attrape les bugs de client tôt.

🏋️ Exercices

Exercice 1 — Le poste-frontière (implémenter)

Objectif. Écris un modèle WebhookEvent qui valide un payload de webhook : un champ event_type parmi un énum (created, updated, deleted), un timestamp (accepte un epoch int ou une str ISO), un payload: dict[str, object]. Les payloads invalides doivent lever une ValidationError exploitable en 422.

Indice/Solution. Utilise Literal["created", "updated", "deleted"] pour l'énum (plus simple qu'un Enum ici). Pour le timestamp polymorphe, déclare timestamp: datetime — Pydantic coerce un epoch int et une str ISO en datetime nativement. Vérifie avec deux entrées (1718530000 et "2024-06-16T10:00:00Z") qu'elles produisent le même type.

Exercice 2 — Type contraint réutilisable (production-grade)

Objectif. Crée un type MonetaryAmount = Annotated[Decimal, ...] qui : refuse les valeurs négatives, force exactement 2 décimales, et se sérialise en str en JSON (pas en float, pour éviter les erreurs d'arrondi). Utilise-le dans un modèle Invoice avec total: MonetaryAmount.

Indice/Solution. Decimal + Field(ge=0, max_digits=..., decimal_places=2). Pour la sérialisation en str, ajoute un field_serializer ou configure model_config = ConfigDict(json_encoders=...) — en v2 préfère Annotated[Decimal, PlainSerializer(lambda d: str(d), return_type=str)]. Teste que Invoice(total=Decimal("10.5")).model_dump_json() donne "10.50" en string.

Exercice 3 — Validation des tool inputs d'un agent (production-grade)

Objectif. Reprends la boucle tool-use ci-dessus. Ajoute un second tool search_db(query: str, limit: int) avec limit borné entre 1 et 100. Force le modèle à appeler un tool avec des arguments hors bornes (par un prompt adverse), et vérifie que ta boucle renvoie is_error: True avec les errors() au lieu d'exécuter le tool. Confirme que le modèle se corrige au tour suivant.

Indice/Solution. SearchArgs avec limit: int = Field(ge=1, le=100). Le model_json_schema() ne transmettra peut-être pas les bornes via structured outputs côté tool, donc c'est ton model_validate(block.input) qui doit attraper limit=9999. Le renvoi de exc.errors() dans le tool_result donne au modèle le type: "less_than_equal" — il comprend et réessaie. C'est le pattern de réparation autonome.

Exercice 4 — Casser puis réparer : la coercion silencieuse (break-then-fix)

Objectif. Crée un modèle Config avec replicas: int et enabled: bool. Construis-le avec Config(replicas="3", enabled="false"). Observe ce que vaut enabled. (Spoiler : "false" non-vide coerce en True dans certaines versions/réglages, ou suit des règles surprenantes.) Puis répare en passant le champ critique en strict.

Indice/Solution. En v2, bool a des règles de coercion précises ("false", "0", "no" → False ; "true", "1" → True ; une str arbitraire lève). Le vrai piège est replicas="3" qui passe silencieusement. La réparation : replicas: Annotated[int, Field(strict=True)] rejette "3" et force l'appelant à passer un vrai int. Discute du tradeoff : strict partout casse la coercion pratique des bodies HTTP ; strict ciblé sur les champs sensibles est le bon compromis.

Exercice 5 — Perf : valider 100k lignes (production-grade)

Objectif. Tu reçois list[dict] de 100 000 lignes à valider en un modèle Row. Compare deux approches : (a) [Row.model_validate(r) for r in rows] et (b) TypeAdapter(list[Row]).validate_python(rows). Mesure avec time.perf_counter().

Indice/Solution. Le TypeAdapter valide la collection en un seul passage dans le cœur Rust, sans repayer l'overhead de la boucle Python ni la résolution de schéma par item. Tu devrais voir un facteur 2 à 5×. Leçon senior : sur les chemins chauds, valide en batch, pas en boucle. Bonus : si une seule ligne casse tout le batch, attrape ValidationError et lis exc.errors()[i]["loc"] pour savoir quelle ligne — le loc contient l'index.

🎤 En entretien

Q : Quelle est la différence entre une annotation de type Python et la validation Pydantic ? Une annotation (age: int) est ignorée au runtime — c'est de la doc pour mypy/l'IDE, comme un type TS effacé à la compilation. Pydantic fait respecter l'annotation au runtime à la construction du BaseModel : il valide, coerce et lève une ValidationError si la donnée ne correspond pas.

Q : Pourquoi séparer modèle de requête et modèle de réponse dans une API FastAPI ? C'est une frontière de sécurité et de contrat : le modèle de réponse agit comme un filtre de sérialisation qui garantit qu'aucun champ sensible (hash de mot de passe, notes internes) ne fuit, même si l'objet sous-jacent les contient — et il découple le schéma exposé de ton schéma interne, donc un refactor DB ne change pas ton API publique.

Q : Comment fiabilises-tu une réponse LLM en un objet typé ? J'utilise les structured outputs : je passe un modèle Pydantic à client.messages.parse() (output_config format), le SDK contraint la sortie au JSON Schema et me renvoie une instance déjà validée via parsed_output — fini le json.loads + try/except sur du texte. Et pour les arguments de tools, je valide block.input avec model_validate avant d'exécuter, en renvoyant is_error + les errors() au modèle pour qu'il s'auto-corrige.

Q : model_validate vs TypeAdapter, quand utiliser quoi ?model_validate pour un objet unique. Pour valider une collection homogène sur un chemin chaud, TypeAdapter(list[Model]).validate_python(...) valide tout en un passage dans pydantic-core (Rust) et évite l'overhead de la boucle Python — typiquement 2 à 5× plus rapide. TypeAdapter sert aussi à valider des types qui ne sont pas des BaseModel (un list[int], un dict[str, Decimal]).