Pydantic v2 in the Real World: FastAPI, Settings, Pipelines

Practical Pydantic v2 patterns: FastAPI integration, settings, ETL pipelines, API clients, and database models

TL;DR — Key Takeaways

• Use split models (BookCreate / BookRead / BookUpdate) in FastAPI to keep contracts honest and prevent accidental secret leaks.
• pydantic-settings gives your config the same validation guarantees as your data models — fail fast on startup, not in production.
• Validate at the pipeline boundary: collect all errors per record rather than aborting on the first failure.
• Wrap HTTP clients with Pydantic response models; parse failures surface immediately at the deserialization point.
• Keep SQLAlchemy ORM models and Pydantic models separate; bridge them with from_attributes=True.
• SecretStr masks secrets in repr() and logs — it does not encrypt. Call .get_secret_value() when you need the underlying string.

Contents

• FastAPI Integration
• Configuration Management
• Data Pipelines and ETL
• Type-Safe API Client Wrappers
• Database ORM Integration
• Structured LLM Output with Pydantic
• FAQ

You have learned how Pydantic models work, how validators and custom types give you fine-grained control, and how serialization lets you shape output precisely (see Part 2: Core Tutorial and Part 3: Advanced Features). Now it is time to zoom out and look at how Pydantic fits into the systems you actually build every day.

This post is a field guide. Each section covers a distinct domain — web APIs, application configuration, data pipelines, API client wrappers, and database integration — and shows you not just what the code looks like, but why the pattern is worth adopting. By the end, you will have a toolkit of repeatable, production-tested patterns you can drop into your next project.

FastAPI Integration

FastAPI and Pydantic were designed together, and it shows. FastAPI uses Pydantic models to declare request bodies, validate query parameters, and serialize response payloads. Understanding the mechanics of that relationship makes you a far more effective FastAPI developer.

If FastAPI is new to you, this section assumes basic familiarity — see the FastAPI tutorial for an introduction.

Here is the simplest possible FastAPI app to orient you:

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
def health() -> dict:
    return {"status": "ok"}

Everything below builds on this foundation.

How FastAPI Uses Pydantic Under the Hood

When you annotate a path operation function parameter with a Pydantic model, FastAPI:

1. Reads the raw request body (JSON bytes).
2. Calls model.model_validate_json(body) on your model class (for JSON body parameters; query and path parameters follow separate validation paths).
3. Raises an HTTP 422 Unprocessable Entity if ValidationError is thrown, serializing the error detail automatically.
4. Passes the validated model instance to your function.

On the way out, when you declare response_model=SomeModel, FastAPI calls model.model_dump() on whatever your function returns, then serializes that dict to JSON. This means your function can return an ORM object, a raw dict, or a model instance — FastAPI handles the conversion as long as the data is compatible.

The Split Model Pattern

A common mistake in FastAPI projects is using a single Pydantic model for every operation: creation, reads, updates, and responses. This leads to awkward Optional fields everywhere and security leaks (returning a password_hash because you forgot to exclude it).

The idiomatic approach is to define a shared base and derive specialized models from it:

• BookBase — fields shared across all operations (title, author, isbn).
• BookCreate — adds fields only needed at creation time.
• BookUpdate — makes all fields optional for partial updates.
• BookRead — adds server-generated fields (id, created_at) and excludes secrets.

This is not boilerplate for its own sake. Each model encodes a contract. BookCreate says “here is what the client must provide.” BookRead says “here is what the server guarantees to return.” Keeping them separate keeps those contracts honest.

response_model_exclude_unset

When handling partial updates (PATCH requests), you want to know which fields the client actually sent. Calling model.model_dump(exclude_unset=True) gives you only the fields the client provided — not the defaults. This is essential for applying partial updates correctly without accidentally overwriting existing data with default values.

Note: exclude_unset is shallow. Nested models still emit their full structure. For partial nested updates, dump and merge manually.

Complete Example: Book CRUD API

The constraints on BookBase fields (min_length, max_length, pattern) are extracted into module-level Annotated type aliases so that BookUpdatecan reuse them without duplicating the definitions. This avoids the duplication — both classes reference the same constraint. See Part 3: Advanced Features — reusable constrained types for more on this pattern.

In Pydantic v2, a type annotation without a default is required — no need for Field(...).

from __future__ import annotations

from datetime import datetime, timezone
from typing import Annotated, Any
from uuid import UUID, uuid4
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

# --- Reusable constrained types ---
BookTitle = Annotated[str, Field(min_length=1, max_length=200)]
BookAuthor = Annotated[str, Field(min_length=1, max_length=100)]
BookIsbn = Annotated[str, Field(pattern=r"^\d{13}$")]
BookYear = Annotated[int, Field(ge=1450, le=2100)]

# --- Models ---
class BookBase(BaseModel):
    title: BookTitle
    author: BookAuthor
    isbn: BookIsbn
    published_year: BookYear

class BookCreate(BookBase):
    pass

class BookUpdate(BaseModel):
    title: BookTitle | None = None
    author: BookAuthor | None = None
    isbn: BookIsbn | None = None
    published_year: BookYear | None = None

class BookRead(BookBase):
    id: UUID
    created_at: datetime
    updated_at: datetime
    model_config = {"from_attributes": True}

# --- In-memory store (stand-in for a real database) ---
_books: dict[UUID, dict[str, Any]] = {}

# --- App ---
app = FastAPI(title="Book API")

@app.post("/books", response_model=BookRead, status_code=201)
def create_book(payload: BookCreate) -> dict[str, Any]:
    now = datetime.now(timezone.utc)
    book_id = uuid4()
    record = {
        **payload.model_dump(),
        "id": book_id,
        "created_at": now,
        "updated_at": now,
    }
    _books[book_id] = record
    return record

@app.get("/books/{book_id}", response_model=BookRead)
def get_book(book_id: UUID) -> dict[str, Any]:
    book = _books.get(book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return book

@app.patch("/books/{book_id}", response_model=BookRead)
def update_book(book_id: UUID, payload: BookUpdate) -> dict[str, Any]:
    book = _books.get(book_id)
    if book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    # Only apply fields the client actually sent
    updates = payload.model_dump(exclude_unset=True)
    book.update({**updates, "updated_at": datetime.now(timezone.utc)})
    return book

BookUpdate does not extend BookBase — every field is independently optional. If it did extend BookBase, you would have to override each field to make it optional. By sharing the Annotated type aliases, both classes enforce the same constraints without any duplication.

Validation Errors Become 422 Automatically

If a client POSTs {"title": "", "author": "Orwell", "isbn": "bad", "published_year": 1984}, FastAPI catches the ValidationErrorand returns:

{
  "detail": [
    {"loc": ["body", "title"], "msg": "String should have at least 1 character", "type": "string_too_short"},
    {"loc": ["body", "isbn"], "msg": "String should match pattern '^\\d{13}$'", "type": "string_pattern_mismatch"}
  ]
}

You get structured, machine-readable validation errors for free. Your clients can display field-level error messages without any extra work on your end.

Configuration Management

Application configuration is one of the most error-prone areas of any codebase. Environment variables are stringly-typed by nature, secrets are scattered across .env files and Kubernetes secrets, and bugs caused by a misconfigured DATABASE_URL only surface at runtime — often in production.

pydantic-settings solves this by giving your configuration the same validation guarantees your data models have.

The Singleton Pattern

Configuration should be loaded once and reused everywhere. The typical pattern is a module-level singleton:

from functools import lru_cache

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()

lru_cache ensures Settings() is only instantiated once per process. In tests, you can clear the cache with get_settings.cache_clear() and inject a test configuration.

Multi-Environment Configuration

pydantic-settings reads from environment variables by default, but you can point it at .env files. For multi-environment setups, load the right file based on an APP_ENV variable.

Docker and Kubernetes Secrets

K8s secrets are typically mounted as files under /run/secrets/. pydantic-settings supports this with secrets_dir. When a field named database_password is not found in the environment, pydantic-settings will look for a file at /run/secrets/database_password and read its contents as the value.

Operational note: pydantic-settings reads files in secrets_dir once at instantiation. If your platform rotates secrets in place, you will need to re-instantiate the settings (e.g., on SIGHUP) — Pydantic does not watch the filesystem.

Complete Example: Production-Ready Settings

SecretStr masks the secret in repr(), logs, and exception messages — it does NOT encrypt the value. Call .get_secret_value() to read the underlying string. For example:

print(settings)
# app_name='MyService' database_url=SecretStr('**********') ...

print(settings.database_url.get_secret_value())
# postgresql://user:pass@localhost/mydb

from __future__ import annotations

import os
from functools import lru_cache
from typing import Literal
from pydantic import Field, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=f".env.{os.getenv('APP_ENV', 'dev')}",
        env_file_encoding="utf-8",
        secrets_dir="/run/secrets",
        case_sensitive=False,
    )
    # Application
    app_name: str = "MyService"
    app_env: Literal["dev", "staging", "prod"] = "dev"
    debug: bool = False
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
    # Database
    database_url: SecretStr = Field(description="PostgreSQL connection URL")
    database_pool_size: int = Field(10, ge=1, le=100)
    database_pool_timeout: int = Field(30, ge=5)
    # Redis
    redis_url: str = Field("redis://localhost:6379/0")
    redis_max_connections: int = Field(20, ge=1)
    # External API keys
    stripe_api_key: SecretStr | None = None
    sendgrid_api_key: SecretStr | None = None
    # Feature flags
    enable_new_checkout: bool = False
    enable_beta_dashboard: bool = False
    # CORS
    cors_origins: list[str] = Field(default_factory=list)
    cors_allow_credentials: bool = True
    @field_validator("cors_origins", mode="before")
    @classmethod
    def parse_cors_origins(cls, v: str | list[str]) -> list[str]:
        if isinstance(v, str):
            return [origin.strip() for origin in v.split(",") if origin.strip()]
        return v
    @field_validator("database_url", mode="before")
    @classmethod
    def validate_database_url(cls, v: object) -> object:
        # mode='before' receives the raw pre-coercion input - handle SecretStr explicitly
        if isinstance(v, SecretStr):
            v = v.get_secret_value()
        if not isinstance(v, str) or not v.startswith(("postgresql://", "postgresql+asyncpg://")):
            raise ValueError("database_url must be a PostgreSQL URL")
        return v

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()

The key behaviors here:

• SecretStr prevents credentials from appearing in logs or repr output. Call .get_secret_value() when you actually need the string.
• Literal["dev", "staging", "prod"] on app_env means you cannot accidentally set APP_ENV=developemnt and have it silently pass.
• cors_origins accepts either a comma-separated string (convenient in .env files) or a list (convenient in Python tests).
• The database URL validator provides a clear error message instead of a confusing SQLAlchemy connection error at startup.
• model_config is placed at the top of the class body — this is the convention for BaseSettings subclasses.

Fail-fast is the philosophy here. If your configuration is wrong, you want the process to crash immediately on startup with a clear error, not fail silently three minutes into handling production traffic.

Data Pipelines and ETL

Data pipelines are where Pydantic earns its reputation as a reliability tool. When you are ingesting CSV files, API responses, or messages from a queue, the incoming data is untrusted by definition. Validating at the boundary — before the data enters your system — is the difference between catching errors early and debugging corrupted database records weeks later.

The Core Principle: Validate at the Boundary

Every place where data enters your system from the outside world is a boundary: reading a CSV row, deserializing a Kafka message, calling a third-party API. Validation should happen immediately at that boundary, not somewhere downstream after you have already started processing.

TypeAdapter for Bulk Validation

For validating sequences of items without a wrapper model, TypeAdapter is the right tool:

from pydantic import TypeAdapter

adapter = TypeAdapter(list[UserRecord])
records = adapter.validate_python(raw_data)

ConfigDict(extra='ignore') for Lenient Parsing

When consuming external data sources — especially third-party APIs or legacy CSVs — the source often includes fields you do not care about. Using extra='ignore' prevents ValidationError on unexpected fields, while still validating the fields you do declare.

Error Collection Pattern

In a pipeline, you rarely want to abort on the first invalid row. Instead, collect all errors alongside valid records, then decide what to do: log them, write them to a dead-letter queue, alert a human, or skip them.

Complete Example: CSV User Record Pipeline

The validate_age validator uses info.context to receive an injected reference date so that tests can pass a fixed date instead of relying on date.today(). Pass {"today": date(2025, 1, 1)} (or any fixed date) as the context argument to model_validate; if no context is provided it falls back to the real current date.

Note: EmailStr requires pip install "pydantic[email]" for the email-validator backend.

from __future__ import annotations

import csv
from dataclasses import dataclass, field
from datetime import date
from io import StringIO
from typing import Any
from pydantic import BaseModel, ConfigDict, EmailStr, Field, ValidationError, ValidationInfo, model_validator

class UserRecord(BaseModel):
    model_config = ConfigDict(extra="ignore")
    user_id: int = Field(gt=0)
    full_name: str = Field(min_length=1, max_length=200)
    email: EmailStr
    birth_date: date
    monthly_spend: float = Field(ge=0.0)
    country_code: str = Field(min_length=2, max_length=2)
    @model_validator(mode="after")
    def validate_age(self, info: ValidationInfo) -> UserRecord:
        today: date = (info.context or {}).get("today", date.today())
        age = (today - self.birth_date).days / 365.25
        if age < 18:
            raise ValueError("User must be at least 18 years old")
        return self

@dataclass
class PipelineResult:
    valid: list[UserRecord] = field(default_factory=list)
    invalid: list[dict[str, Any]] = field(default_factory=list)
    @property
    def success_rate(self) -> float:
        total = len(self.valid) + len(self.invalid)
        return len(self.valid) / total if total > 0 else 0.0

def process_user_csv(csv_content: str, today: date | None = None) -> PipelineResult:
    result = PipelineResult()
    reader = csv.DictReader(StringIO(csv_content))
    context = {"today": today} if today is not None else {}
    for row_number, raw_row in enumerate(reader, start=2):  # start=2: row 1 is header
        try:
            record = UserRecord.model_validate(raw_row, context=context)
            result.valid.append(record)
        except ValidationError as exc:
            result.invalid.append({
                "row_number": row_number,
                "raw_data": dict(raw_row),
                "errors": exc.errors(include_url=False),
            })
    return result

# --- Example usage ---
SAMPLE_CSV = """\
user_id,full_name,email,birth_date,monthly_spend,country_code,legacy_field
1,Alice Martin,alice@example.com,1990-03-15,250.00,US,ignored_value
2,Bob Chen,bob@example.com,2010-07-22,50.00,CA,ignored_value
3,Carol Davis,not-an-email,1985-11-01,-10.00,GB,ignored_value
4,Dan Lee,dan@example.com,1978-05-30,0.00,DE,ignored_value
"""
if __name__ == "__main__":
    result = process_user_csv(SAMPLE_CSV, today=date(2026, 4, 29))
    print(f"Valid records: {len(result.valid)}")
    print(f"Invalid records: {len(result.invalid)}")
    print(f"Success rate: {result.success_rate:.1%}")
    for failure in result.invalid:
        print(f"\nRow {failure['row_number']}: {failure['raw_data']['full_name']}")
        for error in failure["errors"]:
            print(f"  - {'.'.join(str(loc) for loc in error['loc'])}: {error['msg']}")

Running this produces:

Valid records: 2
Invalid records: 2
Success rate: 50.0%

Row 3: Bob Chen
  - : Value error, User must be at least 18 years old
Row 4: Carol Davis
  - email: value is not a valid email address. An email address must have an @-sign.
  - monthly_spend: Input should be greater than or equal to 0

Note that legacy_field in the CSV is silently ignored because of extra="ignore". Bob fails the age validator (born 2010, age 15 as of April 2026). Carol fails two field-level validators, and both errors are collected in a single pass — Pydantic does not stop at the first error within a record.

Type-Safe API Client Wrappers

Calling a REST API and getting back a raw dict or response.json() is one of the biggest sources of subtle bugs in Python code. A key is renamed upstream, a field changes from int to string, a nested object goes nullable — and your code fails in a completely unrelated place, minutes or hours later.

Wrapping your HTTP client with Pydantic response models solves this class of problem entirely. The failure happens immediately, at the deserialization point, with a clear message about exactly which field was wrong.

model_validate_json Over model_validate

When you have an HTTP response, prefer Model.model_validate_json(response.content) over Model.model_validate(response.json()). The former parses JSON bytes in a single pass inside Pydantic’s Rust core and is meaningfully faster for large payloads.

Handling camelCase APIs

Many REST APIs return camelCase JSON (userName, createdAt). Your Python models should use snake_case. Pydantic’s alias_generator=to_camel (from pydantic.alias_generators) handles the entire model at once without per-field Field(alias=...). The GitHub API happens to use snake_case, so the example below does not need an alias generator — but the pattern applies wherever you encounter camelCase responses.

Complete Example: Type-Safe GitHub API Client

TypeAdapter is constructed at module scope to avoid per-call overhead — see Part 5: Tips and Best Practices for measured impact.

from __future__ import annotations

from datetime import datetime, timezone
from typing import Generic, TypeVar
import httpx
from pydantic import BaseModel, ConfigDict, TypeAdapter
T = TypeVar("T")

# --- Response models ---
# GitHub's API uses snake_case, so field names match Python conventions directly.
# For camelCase APIs you'd also import `to_camel` and add a config:
#   from pydantic.alias_generators import to_camel
#   model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
class GitHubRepo(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: int
    name: str
    full_name: str
    private: bool
    description: str | None
    html_url: str
    stargazers_count: int
    forks_count: int
    open_issues_count: int
    created_at: datetime
    updated_at: datetime
    pushed_at: datetime | None
    language: str | None
    default_branch: str

class GitHubUser(BaseModel):
    model_config = ConfigDict(extra="ignore")
    id: int
    login: str
    name: str | None
    email: str | None
    public_repos: int
    followers: int
    following: int
    created_at: datetime

# Module-level adapter - constructing TypeAdapter once avoids per-call schema-build overhead
_repo_list_adapter: TypeAdapter[list[GitHubRepo]] = TypeAdapter(list[GitHubRepo])

# --- Client ---
class GitHubAPIError(Exception):
    def __init__(self, status_code: int, message: str) -> None:
        self.status_code = status_code
        super().__init__(f"GitHub API error {status_code}: {message}")

class GitHubClient:
    BASE_URL = "https://api.github.com"
    def __init__(self, token: str | None = None) -> None:
        headers = {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }
        if token:
            headers["Authorization"] = f"Bearer {token}"
        self._client = httpx.Client(base_url=self.BASE_URL, headers=headers)
    def _get(self, path: str, **params: str | int) -> httpx.Response:
        response = self._client.get(path, params=params)
        if response.status_code >= 400:
            raise GitHubAPIError(
                status_code=response.status_code,
                message=response.json().get("message", "Unknown error"),
            )
        return response
    def get_user(self, username: str) -> GitHubUser:
        response = self._get(f"/users/{username}")
        return GitHubUser.model_validate_json(response.content)
    def list_repos(
        self,
        username: str,
        per_page: int = 30,
        page: int = 1,
    ) -> list[GitHubRepo]:
        response = self._get(
            f"/users/{username}/repos",
            per_page=per_page,
            page=page,
            sort="updated",
        )
        return _repo_list_adapter.validate_json(response.content)
    def get_repo(self, owner: str, repo: str) -> GitHubRepo:
        response = self._get(f"/repos/{owner}/{repo}")
        return GitHubRepo.model_validate_json(response.content)
    def close(self) -> None:
        self._client.close()
    def __enter__(self) -> GitHubClient:
        return self
    def __exit__(self, *args: object) -> None:
        self.close()

# --- Example usage ---
if __name__ == "__main__":
    with GitHubClient() as client:
        user = client.get_user("torvalds")
        print(f"{user.login} has {user.public_repos} public repos")
        repos = client.list_repos("torvalds", per_page=5)
        for repo in repos:
            print(f"  {repo.name}: {repo.stargazers_count} stars ({repo.language})")

The extra="ignore" on the response models is important — GitHub’s API returns many more fields than we declare. By ignoring extras, we only validate the fields we care about, making the client resilient to new fields GitHub adds over time.

When model_validate_json(response.content) raises a ValidationError, you know immediately which field the API changed on you and exactly what value was received. Compare that to a KeyError or AttributeError three call frames downstream.

For APIs that use camelCase, add model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True) to your response models. The alias_generator maps Python snake_case names to camelCase aliases, and populate_by_name=True lets you construct models using either naming convention — useful in tests.

Database ORM Integration

Pydantic models describe data shapes. Database ORM models describe storage. These are related concerns but distinct ones, and the way you bridge them has significant implications for your codebase’s flexibility and testability.

SQLModel: Unified Models

SQLModel is a library by the FastAPI author that merges SQLAlchemy table definitions and Pydantic models into a single class:

from sqlmodel import SQLModel, Field  # Note: this is sqlmodel.Field, not pydantic.Field
                                      # It adds DB-specific arguments like primary_key, foreign_key

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None

The trade-offs are real, though. SQLModel’s unified model means your database schema and your API contract are coupled. When you need a HeroReadthat excludes secret_name or a HeroCreate that excludes id, you end up with multiple SQLModel classes anyway — losing much of the simplicity. For greenfield projects with straightforward schemas, it is a reasonable choice. For anything complex, the manual approach gives you cleaner separation.

The Manual Approach: Separate Models, Explicit Conversion

The more scalable pattern is to keep your SQLAlchemy ORM models and your Pydantic models entirely separate and write explicit conversion functions. The key Pydantic config for ORM integration is from_attributes=True. Without it, model_validate(orm_instance) raises an error because ORM instances are not dicts. With it, Pydantic reads attributes off the object directly.

Complete Example: SQLAlchemy + Pydantic Side by Side

from __future__ import annotations

from datetime import datetime
from typing import Any
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

# --- SQLAlchemy ORM layer ---
class Base(DeclarativeBase):
    pass

class UserORM(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now(), nullable=False
    )

# --- Pydantic API layer ---
class UserBase(BaseModel):
    username: str = Field(
        min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$"
    )
    email: EmailStr

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

class UserUpdate(BaseModel):
    username: str | None = Field(
        None, min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$"
    )
    email: EmailStr | None = None
    is_active: bool | None = None

class UserRead(UserBase):
    model_config = ConfigDict(from_attributes=True)
    id: int
    is_active: bool
    created_at: datetime
    updated_at: datetime

# --- Conversion utilities ---
def hash_password(password: str) -> str:
    # Placeholder - in production, use bcrypt or argon2:
    #   import bcrypt
    #   return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
    return f"hashed_{password}"

def user_orm_to_read(orm_user: UserORM) -> UserRead:
    return UserRead.model_validate(orm_user)

def user_create_to_orm(payload: UserCreate) -> UserORM:
    return UserORM(
        username=payload.username,
        email=payload.email,
        hashed_password=hash_password(payload.password),
    )

# --- Repository (thin data-access layer) ---
class UserRepository:
    def __init__(self, session: Session) -> None:
        self._session = session
    def create(self, payload: UserCreate) -> UserRead:
        orm_user = user_create_to_orm(payload)
        self._session.add(orm_user)
        self._session.flush()
        self._session.refresh(orm_user)
        return user_orm_to_read(orm_user)
    def get_by_id(self, user_id: int) -> UserRead | None:
        orm_user = self._session.get(UserORM, user_id)
        if orm_user is None:
            return None
        return user_orm_to_read(orm_user)
    def update(self, user_id: int, payload: UserUpdate) -> UserRead | None:
        orm_user = self._session.get(UserORM, user_id)
        if orm_user is None:
            return None
        updates = payload.model_dump(exclude_unset=True)
        for key, value in updates.items():
            setattr(orm_user, key, value)
        self._session.flush()
        self._session.refresh(orm_user)
        return user_orm_to_read(orm_user)
    def list_active(self) -> list[UserRead]:
        from sqlalchemy import select
        stmt = select(UserORM).where(UserORM.is_active.is_(True))
        orm_users = self._session.scalars(stmt).all()
        return [user_orm_to_read(u) for u in orm_users]

A few things worth noting in this design:

from_attributes=True is only on UserRead. The other models do not need it because they are never populated from ORM instances.

model_dump(exclude_unset=True) in the update method is essential. If a client sends {"is_active": false}, you want to set only is_active. Without exclude_unset=True, you would also apply None to username and email, which would overwrite existing data with null values.

Security note: blindly setattr-ing user input is safe here only because UserUpdate is a closed schema. If you broaden the schema or derive the update dict from a different model, you risk mass-assignment vulnerabilities — always validate against a schema whose surface you control, never the raw request body directly.

The hashed_password field exists on UserORM but not on UserRead. This is the split model pattern’s security benefit in action: no amount of serialization misconfiguration can leak the password hash through UserRead, because the field simply does not exist on the model.

Structured LLM Output with Pydantic

One of the most exciting recent applications of Pydantic is constraining large language model outputs to well-defined structures. When you ask an LLM to extract information or reason over data, you want a predictable, validated Python object — not a string you have to parse yourself.

pydantic-ai is a framework that uses Pydantic models as the return type contract for LLM calls:

from __future__ import annotations

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class BookRecommendation(BaseModel):
    title: str
    author: str
    year: int = Field(ge=1000, le=2100)
    reason: str = Field(min_length=20)
    genre: str

# Agent[system_deps_type, result_type]
agent: Agent[None, BookRecommendation] = Agent(
    "openai:gpt-4o",
    result_type=BookRecommendation,
    system_prompt="You are a literary expert. Recommend one book based on the user's input.",
)

async def get_recommendation(user_input: str) -> BookRecommendation:
    result = await agent.run(user_input)
    return result.data  # Fully validated BookRecommendation instance

Agent[None, BookRecommendation] is a generic type: the first parameter is the type of system dependencies injected at runtime (none here), and the second is the result type Pydantic validates the LLM output against. result.data is a fully validated BookRecommendation instance.

The result_type=BookRecommendation tells pydantic-ai to instruct the model to produce output matching that schema and to validate the result with Pydantic before returning it. If the LLM produces output that fails validation — a year of "nineteen eighty-four" instead of 1984, say — pydantic-ai can retry the call automatically.

This pattern extends naturally to any structured extraction task: parsing invoices, classifying support tickets, extracting entities from documents. You define the expected output shape as a Pydantic model, and the validation layer ensures what you get back is actually usable.

Wrapping Up

You have now seen Pydantic operating across five distinct domains: request/response validation in FastAPI, type-safe configuration loading, pipeline boundary enforcement, HTTP client response modeling, and ORM integration. The through-line in every case is the same: push validation to the edges of your system, make contracts explicit in code, and let Pydantic surface violations immediately rather than letting bad data propagate silently.

These patterns are not theoretical. They are the patterns you will find in production Python services at companies shipping code today.

FAQ

How do I use Pydantic with FastAPI?

Annotate your path operation function parameters with Pydantic models. FastAPI automatically validates incoming JSON, returns 422 on validation errors, and serializes response models. See the FastAPI Integration section above.

What is the difference between request and response models?

Request models (BookCreate, BookUpdate) define what a client must send. Response models (BookRead) define what the server guarantees to return. Keeping them separate prevents accidentally exposing server-only fields (like hashed_password) and avoids awkward Optional fields that exist only for one operation.

How do I load secrets from environment variables?

Use pydantic-settings with SecretStr fields. SecretStr masks the value in repr() and logs. The value is read from the environment at instantiation and is accessible via .get_secret_value(). For Kubernetes, point secrets_dir at your mounted secrets path.

Should I use SQLModel or separate ORM/Pydantic models?

For simple schemas, SQLModel is convenient. For anything complex — multiple read/write shapes, security boundaries between API and DB layers, or independent evolution of schema and API — keep SQLAlchemy ORM models and Pydantic models separate. See the Database ORM Integration section for the trade-offs.

Next in the Series

In Part 5: Pydantic v2 Best Practices, Performance, and Migration, we cover the lessons that come from experience: performance optimization, common pitfalls, testing strategies, migration from v1 to v2, and project architecture patterns. Then Part 6: Pydantic v2 Capstone — Building a Small Service brings the whole series together by building a working bookmarks API end-to-end.

See you there