Validation with Pydantic v2 and JSONB: typed JSONB

Capsule description

JSONB without validation is dict[str, Any] — maximum flexibility, zero guarantees. Your endpoint accepts anything, your database stores anything, and the bug shows up three weeks later in production when a client sends {"og_image": 12345} and the frontend tries to render the URL as a string.

Pydantic v2 solves this. It lets you declare the expected shape of the JSONB with type hints, validate it when it arrives at the endpoint, normalize values when you save it, and keep JSONB's flexibility for the cases where you really do need extensibility. It's the difference between "JSONB as a bag of anything" and "JSONB as structured data with a flexible schema."

This capsule teaches you the three patterns that cover 95% of cases: validating a JSONB sub-object with a Pydantic model, validating polymorphic payloads with discriminated unions (an event_type that decides the schema), and validating on the way out (returning the same Pydantic model to the client so your API has a contract). You'll also learn TypeAdapter for validating Python dicts without going through FastAPI.

By the end you'll be able to declare JSONB with a typed shape, receive and return validated data in FastAPI endpoints, and apply discriminated unions when your JSONB changes schema based on a discriminating key.

Mental model: two worlds meeting

You have two representations of the same data:

┌─────────────────────────────┐         ┌───────────────────────────────┐
│  PostgreSQL                 │         │  Python                       │
│                             │         │                               │
│  posts.metadata JSONB       │ ←──→    │  metadata: PostMetadata       │
│                             │         │  (Pydantic model)             │
│  Flexible storage           │         │  Validation on the way in     │
│  Queries with @>, ->>, etc. │         │  Type safety in code          │
└─────────────────────────────┘         └───────────────────────────────┘
                                                    ↑
                                                    │
                                          ┌──────────────────────┐
                                          │  Three bridges       │
                                          │                      │
                                          │  1. Endpoint in      │
                                          │  2. Persistence      │
                                          │  3. Endpoint out     │
                                          └──────────────────────┘

The mental trick: SQLAlchemy stores and reads Python dicts. Pydantic takes that dict (or JSON) and validates it against a schema. You decide at which boundary to apply the validation.

The three canonical boundaries:

1. Endpoint input. The client sends JSON. Pydantic validates it before touching the SQLAlchemy session. If it fails, FastAPI returns an automatic 422.

2. Persistence. Before saving, you convert the Pydantic model to a dict with .model_dump(mode="json") and assign it to the JSONB column. This normalizes types (datetime → ISO string, Decimal → string, etc.) into JSON-compatible ones.

3. Endpoint output. You load the JSONB as a dict, validate it against the same Pydantic model with .model_validate(), and return it. The client receives data consistent with the declared contract.

If you do all three, your JSONB has a contract. If you only do one, you have a "courtesy validation" that breaks the moment someone writes directly to the database.

Pattern 1: validating a JSONB sub-object with a Pydantic model

The most common case: your metadata column has a seo sub-key with known fields. You want to validate seo when you receive it and return it with the same schema.

# models.py
from typing import Any

from pydantic import BaseModel, Field, HttpUrl
from sqlalchemy import BigInteger
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",
        MutableDict.as_mutable(JSONB),
        nullable=False,
        default=dict,
    )


# Pydantic models
class PostSEO(BaseModel):
    """Sub-schema inside metadata.seo. Known fields, validated."""
    title: str = Field(min_length=1, max_length=160)
    description: str = Field(min_length=1, max_length=320)
    canonical: HttpUrl
    og_image: HttpUrl | None = None


class PostMetadata(BaseModel):
    """Full metadata schema. Allows extensibility for non-validated fields."""
    seo: PostSEO | None = None
    tags: list[str] = Field(default_factory=list)
    published: bool = False
    # Any other field is accepted but not validated, for extensibility:
    model_config = {"extra": "allow"}

Three important details in the Pydantic model:

1. HttpUrl validates that the string is a valid URL (http://... or https://...). If og_image: "i am not a url" arrives, validation fails.

2. Field(min_length=..., max_length=...) validates limits. Useful for SEO (title has a recommended maximum).

3. extra = "allow" permits additional undeclared keys. This preserves JSONB's flexibility: if the team adds metadata.experiment_variant tomorrow without telling you, you don't break the endpoint. Known keys are validated; unknown ones are accepted as-is.

If instead you want to be strict and reject unknown fields, use "extra": "forbid". That's the typical choice for "closed" schemas.

Input endpoint with validation

# api.py
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


engine = create_async_engine(
    "postgresql+asyncpg://postgres:postgres@localhost:5432/demo"
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()


class CreatePostRequest(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    metadata: PostMetadata = Field(default_factory=PostMetadata)


@app.post("/posts", status_code=status.HTTP_201_CREATED)
async def create_post(payload: CreatePostRequest) -> dict[str, Any]:
    """
    FastAPI already validates the body against CreatePostRequest. If it
    fails, it responds 422 with the error details. You don't have to do
    anything extra.
    """
    async with SessionLocal() as session:
        post = Post(
            title=payload.title,
            # Convert the Pydantic model to a JSON-serializable dict
            metadata_=payload.metadata.model_dump(mode="json"),
        )
        session.add(post)
        await session.commit()
        return {
            "id": post.id,
            "title": post.title,
            "metadata": post.metadata_,
        }

Why mode="json":

payload.metadata.model_dump() (with no argument) returns a dict with Python types: HttpUrl is still a Url object, datetime is still a datetime, Decimal is still a Decimal. JSONB needs JSON-serializable types (string, number, bool, null, dict, list).

.model_dump(mode="json") converts them: Url → string, datetime → ISO string, Decimal → string. That's what you need to pass to SQLAlchemy.

Output endpoint with re-validation

class GetPostResponse(BaseModel):
    id: int
    title: str
    metadata: PostMetadata


@app.get("/posts/{post_id}", response_model=GetPostResponse)
async def get_post(post_id: int) -> GetPostResponse:
    async with SessionLocal() as session:
        post = (
            await session.execute(select(Post).where(Post.id == post_id))
        ).scalar_one_or_none()
        if post is None:
            raise HTTPException(404, "post not found")

        # Validate the dict from the database against the schema before returning
        return GetPostResponse(
            id=post.id,
            title=post.title,
            metadata=PostMetadata.model_validate(post.metadata_),
        )

Why validate on the way out too:

• If someone inserted data directly into the database without going through the API (a DBA, a migration script, another app), your endpoint detects the inconsistency and returns an informative 500 instead of handing the client malformed data.
• FastAPI respects the response_model and filters out undeclared fields when serializing — but if you also want to validate types, model_validate guarantees it.

Pattern 2: validating the whole JSONB with TypeAdapter

When you work with SQLAlchemy outside a FastAPI endpoint (jobs, scripts, tests), Pydantic v2's TypeAdapter lets you validate a dict against any type without defining a wrapper model.

from pydantic import TypeAdapter

# Validate a Python dict against PostMetadata without instantiating the model by hand
metadata_adapter = TypeAdapter(PostMetadata)

# In a migration job:
async def fix_metadata_for_post(session, post_id: int, raw_metadata: dict) -> None:
    validated = metadata_adapter.validate_python(raw_metadata)
    post = (await session.execute(select(Post).where(Post.id == post_id))).scalar_one()
    post.metadata_ = validated.model_dump(mode="json")
    await session.commit()

TypeAdapter also works for types without a BaseModel:

TagListAdapter = TypeAdapter(list[str])
tags = TagListAdapter.validate_python(["python", "postgres"])  # OK
TagListAdapter.validate_python(["python", 123])  # ValidationError

Useful for JSONB sub-fields that have a known type but don't deserve a wrapper model.

Pattern 3: discriminated unions for polymorphic payloads

The most interesting case. Your JSONB has a different schema depending on a discriminating key. The classic example: events with a payload that changes based on event_type.

# events.py
from typing import Annotated, Literal

from pydantic import BaseModel, Field
from sqlalchemy import BigInteger
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import Mapped, mapped_column


class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    payload: Mapped[dict[str, Any]] = mapped_column(
        MutableDict.as_mutable(JSONB), nullable=False, default=dict
    )


# Three payload variants
class PurchaseEvent(BaseModel):
    event_type: Literal["purchase"]
    amount: float = Field(gt=0)
    currency: str = Field(min_length=3, max_length=3)
    user_id: int


class SignupEvent(BaseModel):
    event_type: Literal["signup"]
    user_id: int
    referrer: str | None = None
    source: Literal["organic", "paid", "referral"] = "organic"


class ViewEvent(BaseModel):
    event_type: Literal["view"]
    page: str
    user_id: int | None = None  # Anonymous allowed
    duration_ms: int = Field(ge=0)


# Discriminated union
EventPayload = Annotated[
    PurchaseEvent | SignupEvent | ViewEvent,
    Field(discriminator="event_type"),
]

EventPayloadAdapter = TypeAdapter(EventPayload)

How the discriminated union works:

Pydantic looks at the event_type key of the incoming dict. If it's "purchase", it validates against PurchaseEvent. If it's "signup", against SignupEvent. If it's "view", against ViewEvent. If it's anything else, it fails with a clear error: "input type X did not match any discriminator value".

Why it matters:

• Type-specific validation. purchase requires a positive amount and a 3-letter currency; signup has no amount. Without a discriminator, you'd have one schema with every field optional and manual validation in the handler.
• Type narrowing in Python. If your function receives EventPayload and checks if event.event_type == "purchase":, mypy/pyright knows event is a PurchaseEvent and you know all its fields.
• Clear errors for the client. If they send event_type: "purchase" without amount, Pydantic responds with a specific message about what's missing.

Endpoint that receives polymorphic events

@app.post("/events", status_code=status.HTTP_201_CREATED)
async def create_event(payload: EventPayload) -> dict[str, Any]:
    """
    FastAPI validates the body as EventPayload. If event_type doesn't
    match any of the three, it responds 422 before entering the function.
    """
    async with SessionLocal() as session:
        event = Event(payload=payload.model_dump(mode="json"))
        session.add(event)
        await session.commit()
        return {"id": event.id, "payload": event.payload}

Example requests:

# OK
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "event_type": "purchase",
  "amount": 99.99,
  "currency": "USD",
  "user_id": 42
}'

# OK
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "event_type": "view",
  "page": "/home",
  "user_id": 42,
  "duration_ms": 1500
}'

# 422 — amount is missing
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "event_type": "purchase",
  "currency": "USD",
  "user_id": 42
}'
# Response:
# {"detail":[{"type":"missing","loc":["body","amount"],"msg":"Field required",...}]}

# 422 — unknown discriminator
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "event_type": "deletion",
  "user_id": 42
}'
# Response:
# {"detail":[{"type":"union_tag_invalid","loc":["body","event_type"],...}]}

Filtering and validating on read

@app.get("/events/purchases", response_model=list[PurchaseEvent])
async def list_purchases() -> list[PurchaseEvent]:
    async with SessionLocal() as session:
        # Filter by discriminator using the contains you already know
        stmt = select(Event).where(
            Event.payload.contains({"event_type": "purchase"})
        )
        events = (await session.execute(stmt)).scalars().all()

        # Validate each payload against PurchaseEvent specifically
        return [PurchaseEvent.model_validate(e.payload) for e in events]

Event.payload.contains({"event_type": "purchase"}) takes advantage of your GIN from module 1 — a fast query. Then you validate in Python that each payload really is a valid PurchaseEvent (this protects you from old, malformed data).

Worked example: full pipeline in → persistence → out

An end-to-end case. A POST /posts/{id}/metadata endpoint that updates metadata.seo with validated partial fields and returns the full validated JSONB:

# end_to_end.py
import asyncio
from typing import Any

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field, HttpUrl
from sqlalchemy import BigInteger, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",
        MutableDict.as_mutable(JSONB),
        nullable=False,
        default=dict,
    )


class PostSEO(BaseModel):
    title: str = Field(min_length=1, max_length=160)
    description: str = Field(min_length=1, max_length=320)
    canonical: HttpUrl
    og_image: HttpUrl | None = None


class PostMetadata(BaseModel):
    seo: PostSEO | None = None
    tags: list[str] = Field(default_factory=list)
    published: bool = False
    model_config = {"extra": "allow"}


class PatchSEORequest(BaseModel):
    """
    Partial patch: every field optional. Only the ones that come in the
    request get applied.
    """
    title: str | None = Field(default=None, min_length=1, max_length=160)
    description: str | None = Field(default=None, min_length=1, max_length=320)
    canonical: HttpUrl | None = None
    og_image: HttpUrl | None = None


class GetMetadataResponse(BaseModel):
    id: int
    metadata: PostMetadata


engine = create_async_engine(
    "postgresql+asyncpg://postgres:postgres@localhost:5432/demo"
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()


@app.patch(
    "/posts/{post_id}/seo",
    response_model=GetMetadataResponse,
    status_code=status.HTTP_200_OK,
)
async def patch_seo(post_id: int, patch: PatchSEORequest) -> GetMetadataResponse:
    async with SessionLocal() as session:
        post = (
            await session.execute(select(Post).where(Post.id == post_id))
        ).scalar_one_or_none()
        if post is None:
            raise HTTPException(404, "post not found")

        # Validate the current JSONB before merging
        current = PostMetadata.model_validate(post.metadata_)

        # Build the new SEO by merging what exists with the patch
        existing_seo = current.seo.model_dump() if current.seo else {}
        patch_dict = patch.model_dump(exclude_none=True, mode="json")
        merged_seo_dict = {**existing_seo, **patch_dict}

        # Re-validate the full merged SEO (this forces the required fields
        # to be there; if there was only a title and the patch adds a
        # description, the merge must have canonical too)
        try:
            merged_seo = PostSEO.model_validate(merged_seo_dict)
        except Exception:
            raise HTTPException(
                422,
                "Incomplete SEO after the patch. Required fields are missing.",
            )

        # Assign to the full Pydantic model
        current.seo = merged_seo

        # Reassign the root dict (MutableDict detects this)
        post.metadata_["seo"] = merged_seo.model_dump(mode="json")
        await session.commit()

        return GetMetadataResponse(id=post.id, metadata=current)

What's going on:

1. We validate the current JSONB before merging. If the database has malformed data, we fail before propagating it.
2. The patch has every field optional, but merging it with what exists forms a complete PostSEO. If the merge comes out incomplete (missing canonical, for example), Pydantic fails and we return a 422 with an informative message.
3. We assign to the seo sub-key (a root reassignment, which MutableDict detects) and commit.
4. We return the full validated Pydantic model, not the raw dict.

The client always receives consistent data; the database never ends up with malformed SEO.

Why does this matter in real work?

1. Clear, actionable errors for the client. Without Pydantic, a malformed payload reaches your logic and breaks somewhere deep (rendering HTML, consuming the field from another service). With Pydantic, FastAPI responds 422 with the exact list of errors: which field failed, what was expected, what arrived. The frontend knows what to show the user.

2. Type safety in code. If your Python function receives dict[str, Any], mypy/pyright can't help you and you write metadata["seo"]["title"] with your fingers crossed. With PostMetadata, you write metadata.seo.title and the editor autocompletes it — and warns you if seo can be None.

3. Automatic documentation. FastAPI generates the OpenAPI/Swagger spec from your Pydantic models. Your mobile team reads the docs and knows exactly what shape to expect. Without Pydantic, that doc says "metadata: object" — useless.

4. Validation against old data. Validating on the way out also catches problems from incomplete migrations or dirty data. Better an informative 500 in your app than malformed data reaching the client.

5. Discriminated unions avoid the giant if-elif. Without a discriminator, your handler has if event["type"] == "purchase": validate_purchase(event); elif event["type"] == "signup":.... Dozens of lines of manual validation with latent bugs. With a discriminator, Pydantic does it in one declaration.

Traps and common mistakes

Mistake 1 (conceptual): using dict[str, Any] and forgetting to validate

Symptom: your endpoint accepts anything. Production breaks when a client sends something absurd.

Why it happens: the temptation of "JSONB is flexible, we won't validate it" comes fast. It works until real data arrives.

How to fix it: define at least the keys your app reads. The ones it doesn't read can stay as extra="allow". Validating the ones that matter doesn't break flexibility.

Mistake 2 (practical): forgetting mode="json" in model_dump

Symptom: you assign post.metadata_ = my_model.model_dump() and on commit you get an error like Object of type Url is not JSON serializable.

Why it happens: without mode="json", Pydantic's types (Url, datetime, Decimal, UUID) come out as Python objects. JSONB needs JSON types.

Fix: always mode="json" when you're about to persist or serialize to JSON: my_model.model_dump(mode="json").

Mistake 3 (conceptual): declaring every field as Optional

Symptom: your schema accepts any subset of fields. Validation that doesn't validate.

Why it happens: "so I don't break anything, I'll leave it optional." A classic anti-pattern. The Pydantic model ends up as dict[str, Any] with extra steps.

Fix: distinguish required (no default) from optional (with a default or | None). If a field is essential for your app to work, mark it required. The exception is the partial patch (PATCH endpoints), where everything really is optional but you validate that the merged result is complete.

Mistake 4 (practical): a badly declared discriminator

Symptom: the discriminated union doesn't work, Pydantic tries to validate against every variant and returns confusing combined errors.

Why it happens: you forgot the Annotated[..., Field(discriminator="key")], or you didn't use Literal["value"] in each variant.

Fix:

# WRONG
EventPayload = PurchaseEvent | SignupEvent | ViewEvent  # no discriminator

# RIGHT
EventPayload = Annotated[
    PurchaseEvent | SignupEvent | ViewEvent,
    Field(discriminator="event_type"),
]
# And each variant must have:
class PurchaseEvent(BaseModel):
    event_type: Literal["purchase"]  # ← Literal, not str
    ...

Mistake 5 (conceptual): not validating on the way out

Symptom: malformed data in the database reaches the client. Crashes in the frontend.

Why it happens: you validate on the way in and assume the database is always clean. You forget that migrations, manual scripts, or schema changes can dirty it.

Fix: validate on the way out too. MyModel.model_validate(data_from_db) before returning. Whatever is wrong in the database, you catch it and return a 500 with a clear cause, instead of a 200 with broken data.

Mistake 6 (conceptual): confusing extra="allow" with extra="ignore"

Symptom: you declare extra="allow" expecting unknown keys to be persisted, but when you run model_dump they don't show up.

Why it happens: extra="allow" permits the keys to arrive in the input. But model_dump() by default only serializes the declared fields.

Fix: to preserve extra keys on serialization, use model_dump(exclude_unset=True) and make sure model_config = {"extra": "allow"} is set at every level. If you want the extras stored literally in JSONB, better: convert the raw dict from the request instead of going through the full Pydantic model.

class PostMetadata(BaseModel):
    seo: PostSEO | None = None
    tags: list[str] = Field(default_factory=list)
    published: bool = False
    model_config = {"extra": "allow"}

m = PostMetadata.model_validate({"seo": None, "tags": [], "published": True, "experiment": "v3"})
m.model_dump(mode="json")
# Returns: {"seo": None, "tags": [], "published": True, "experiment": "v3"}

With extra="allow", extras are stored in __pydantic_extra__ and model_dump includes them. Confirm it with your tests.

Exercises

Exercise 1: define the Pydantic model

You have a POST /products endpoint that receives a JSON with this structure. Define the Pydantic models you need. Apply reasonable validations.

{
  "name": "T-shirt",
  "price": 29.99,
  "currency": "USD",
  "metadata": {
    "dimensions": {
      "width_cm": 50,
      "height_cm": 70,
      "weight_g": 200
    },
    "materials": ["cotton", "elastane"],
    "available_sizes": ["S", "M", "L", "XL"],
    "in_stock": true
  }
}

from typing import Literal
from pydantic import BaseModel, Field

Size = Literal["XS", "S", "M", "L", "XL", "XXL"]


class ProductDimensions(BaseModel):
    width_cm: float = Field(gt=0)
    height_cm: float = Field(gt=0)
    weight_g: float = Field(gt=0)


class ProductMetadata(BaseModel):
    dimensions: ProductDimensions
    materials: list[str] = Field(min_length=1)
    available_sizes: list[Size] = Field(min_length=1)
    in_stock: bool = True
    model_config = {"extra": "allow"}


class CreateProductRequest(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    price: float = Field(gt=0)
    currency: str = Field(min_length=3, max_length=3, pattern=r"^[A-Z]{3}$")
    metadata: ProductMetadata

Exercise 2: fix an endpoint that doesn't validate

The following endpoint accepts any metadata. Refactor it to use Pydantic.

@app.post("/posts")
async def create_post(payload: dict) -> dict:
    async with SessionLocal() as session:
        post = Post(
            title=payload.get("title", ""),
            metadata_=payload.get("metadata", {}),
        )
        session.add(post)
        await session.commit()
        return {"id": post.id}

from pydantic import BaseModel, Field

class PostSEO(BaseModel):
    title: str = Field(min_length=1, max_length=160)
    description: str | None = None


class PostMetadata(BaseModel):
    seo: PostSEO | None = None
    tags: list[str] = Field(default_factory=list)
    published: bool = False
    model_config = {"extra": "allow"}


class CreatePostRequest(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    metadata: PostMetadata = Field(default_factory=PostMetadata)


@app.post("/posts", status_code=status.HTTP_201_CREATED)
async def create_post(payload: CreatePostRequest) -> dict[str, Any]:
    async with SessionLocal() as session:
        post = Post(
            title=payload.title,
            metadata_=payload.metadata.model_dump(mode="json"),
        )
        session.add(post)
        await session.commit()
        return {"id": post.id, "title": post.title, "metadata": post.metadata_}

Exercise 3: define a discriminated union

Your audit app stores events in audit_log.detail JSONB. There are three event types:

• login: fields user_id (int), ip_address (str), user_agent (str).
• password_change: fields user_id (int), forced_by_admin (bool).
• delete_post: fields user_id (int), post_id (int), reason (str, optional).

Define the discriminated union, the SQLAlchemy model, and a POST /audit-events endpoint that receives and persists them.

from typing import Annotated, Any, Literal

from fastapi import FastAPI, status
from pydantic import BaseModel, Field, IPvAnyAddress, TypeAdapter
from sqlalchemy import BigInteger
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class AuditLog(Base):
    __tablename__ = "audit_log"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    detail: Mapped[dict[str, Any]] = mapped_column(
        MutableDict.as_mutable(JSONB), nullable=False
    )


class LoginEvent(BaseModel):
    event_type: Literal["login"]
    user_id: int
    ip_address: IPvAnyAddress
    user_agent: str = Field(max_length=500)


class PasswordChangeEvent(BaseModel):
    event_type: Literal["password_change"]
    user_id: int
    forced_by_admin: bool = False


class DeletePostEvent(BaseModel):
    event_type: Literal["delete_post"]
    user_id: int
    post_id: int
    reason: str | None = Field(default=None, max_length=500)


AuditDetail = Annotated[
    LoginEvent | PasswordChangeEvent | DeletePostEvent,
    Field(discriminator="event_type"),
]

AuditDetailAdapter = TypeAdapter(AuditDetail)


@app.post("/audit-events", status_code=status.HTTP_201_CREATED)
async def create_audit_event(payload: AuditDetail) -> dict[str, Any]:
    async with SessionLocal() as session:
        event = AuditLog(detail=payload.model_dump(mode="json"))
        session.add(event)
        await session.commit()
        return {"id": event.id, "detail": event.detail}

Exercise 4: validate on the way out

The following endpoint returns raw metadata from the database without validating it. Refactor it to validate before returning, returning a 500 if the data is inconsistent.

@app.get("/posts/{post_id}/metadata")
async def get_metadata(post_id: int) -> dict:
    async with SessionLocal() as session:
        post = (
            await session.execute(select(Post).where(Post.id == post_id))
        ).scalar_one_or_none()
        if post is None:
            raise HTTPException(404, "post not found")
        return post.metadata_

from fastapi import HTTPException
from pydantic import ValidationError


class GetMetadataResponse(BaseModel):
    id: int
    metadata: PostMetadata


@app.get(
    "/posts/{post_id}/metadata",
    response_model=GetMetadataResponse,
)
async def get_metadata(post_id: int) -> GetMetadataResponse:
    async with SessionLocal() as session:
        post = (
            await session.execute(select(Post).where(Post.id == post_id))
        ).scalar_one_or_none()
        if post is None:
            raise HTTPException(404, "post not found")
        try:
            metadata_validated = PostMetadata.model_validate(post.metadata_)
        except ValidationError as exc:
            # Log it so ops knows which post has malformed data
            import logging
            logging.error("malformed metadata in post %d: %s", post_id, exc)
            raise HTTPException(
                500,
                f"metadata for post {post_id} does not match the expected schema",
            )
        return GetMetadataResponse(id=post.id, metadata=metadata_validated)

Exercise 5: TypeAdapter in a batch job

You're writing a job that walks every post and normalizes its tags (lowercase, no duplicates, sorted). metadata.tags must be a list[str]. If a post has malformed tags (not a list, contains non-strings), the job must log it and skip it, not crash.

import asyncio
import logging
from pydantic import TypeAdapter, ValidationError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker

logger = logging.getLogger(__name__)
TagsAdapter = TypeAdapter(list[str])


async def normalize_all_tags(session_factory) -> dict[str, int]:
    """
    Walks every post and normalizes its tags.
    Returns {ok: N, skipped: M}.
    """
    stats = {"ok": 0, "skipped": 0}
    async with session_factory() as session:
        posts = (await session.execute(select(Post))).scalars().all()
        for post in posts:
            raw_tags = post.metadata_.get("tags", [])
            try:
                tags = TagsAdapter.validate_python(raw_tags)
            except ValidationError:
                logger.warning(
                    "post %d: malformed tags (%r), skipping",
                    post.id, raw_tags,
                )
                stats["skipped"] += 1
                continue

            normalized = sorted({t.lower() for t in tags if t})
            if normalized != tags:
                # Root reassignment: MutableDict detects it
                post.metadata_["tags"] = normalized
            stats["ok"] += 1

        await session.commit()
    return stats


# Usage
asyncio.run(normalize_all_tags(SessionLocal))

Summary and next step

In this capsule you learned to type JSONB with Pydantic v2 without sacrificing flexibility:

• Three validation boundaries: input (FastAPI validates automatically), persistence (model_dump(mode="json") for JSON types), output (model_validate before returning).
• extra="allow" preserves JSONB's flexibility when there are undeclared fields you still want to accept.
• TypeAdapter validates any type (not just BaseModels) without wrappers; ideal for jobs and scripts.
• Discriminated unions (Annotated[A | B | C, Field(discriminator="key")]) for polymorphic payloads: variant-specific validation, type narrowing in code, clear errors for the client.
• Validating on the way out too catches malformed data from migrations, direct scripts, and old data.
• mode="json" on persist is non-negotiable when you have Url, datetime, Decimal, or other non-JSON types.

Before moving on you should be able to:

• Define a Pydantic model for JSONB metadata with required fields, optional fields, and allowed extras.
• Write a FastAPI endpoint that receives, validates, persists, and returns metadata with a known shape.
• Apply discriminated unions when a JSONB column changes schema based on a key.
• Recognize when BaseModel beats TypeAdapter and vice versa.

Next capsule — Pattern: dynamic configuration. So far you've seen the how (declaring, mutating, validating). Now you enter the first of the three canonical patterns where JSONB beats relational columns: dynamic configuration. Per-tenant settings, feature flags, per-user customization. Capsule 05 teaches you when JSONB is the right tool for configuration (and when it isn't — because there are cases where a settings table with typed columns is still the right answer).

Resources

1. Pydantic v2 — Models — the official reference, base validation concepts.
2. Pydantic v2 — Discriminated Unions — the official guide to the feature used in pattern 3.
3. Pydantic v2 — TypeAdapter — API reference.
4. Pydantic v2 — model_dump and mode="json" — how to serialize correctly for persistence.
5. FastAPI — Body and validation — Pydantic + FastAPI integration.
6. Sebastián Ramírez (FastAPI) — Path params, query params, body — validation practice in endpoints.
7. Pydantic v2 — Migrating from v1 — useful if your code comes from Pydantic v1, since there were significant changes.

Module 2 — Advanced PostgreSQL for Backend Guide

Next capsule: Pattern: dynamic configuration — per-tenant settings, feature flags, and customization with JSONB.