Validating JSONB documents with Pydantic TypeDecorators

A TypeDecorator over JSONB that calls model_dump() on the way in and model_validate() on the way out gives a JSONB column a real schema: attribute access, defaults in one place, and a ValidationError at the boundary instead of a KeyError three frames later. This guide belongs to querying Postgres JSONB, arrays and full-text search.

Quick Answer

One generic TypeDecorator serves every document model:

from typing import Any, TypeVar

from pydantic import BaseModel
from sqlalchemy import Dialect
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.types import TypeDecorator

M = TypeVar("M", bound=BaseModel)


class PydanticJSONB(TypeDecorator[M]):
    """A JSONB column whose Python value is a validated Pydantic model."""

    impl = JSONB
    cache_ok = True

    def __init__(self, model: type[M], **kwargs: Any) -> None:
        self.model = model
        super().__init__(**kwargs)

    def process_bind_param(self, value: M | dict | None, dialect: Dialect) -> dict | None:
        if value is None:
            return None
        if isinstance(value, dict):
            value = self.model.model_validate(value)   # validate dicts too
        return value.model_dump(mode="json")

    def process_result_value(self, value: dict | None, dialect: Dialect) -> M | None:
        if value is None:
            return None
        return self.model.model_validate(value)

Used on a model:

from pydantic import BaseModel, Field
from sqlalchemy.orm import Mapped, mapped_column

from shop.models import Base


class Shipping(BaseModel):
    carrier: str
    speed: str = "standard"
    signature_required: bool = False
    notes: list[str] = Field(default_factory=list)


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    shipping: Mapped[Shipping] = mapped_column(
        PydanticJSONB(Shipping), default=Shipping(carrier="dhl")
    )

order = Order(shipping=Shipping(carrier="ups", speed="express"))
session.add(order)
await session.commit()

loaded = await session.get(Order, order.id)
loaded.shipping.speed        # "express", typed, not loaded.shipping["speed"]

cache_ok = True matters: without it SQLAlchemy cannot cache statements using the type, and every query recompiles. It is safe here because the type's behaviour depends only on self.model.

mode="json" in model_dump is what turns datetime, Decimal, UUID and Enum values into JSON-representable ones. Without it the driver is handed objects it cannot serialise.

Execution Context & Async Workflow Integration

A TypeDecorator is a pair of translation functions attached to a column type. process_bind_param runs when a value goes to the database; process_result_value runs when a row comes back. Because both live in the type, every part of the application — ORM attributes, Core insert(), select() results, bulk operations — goes through them, with no repeated conversion at call sites.

That uniformity is the real benefit over converting in a service layer. A single place decides what a valid document is, so a bulk insert cannot skip validation, and a raw select(Order.shipping) still returns a model, not a dict.

Async specifics. Both processors are plain synchronous functions, called by the dialect while it prepares parameters or processes rows, and they run inside the greenlet that the async driver uses. So they must not await, and they must not block: no HTTP call to fetch a schema, no synchronous file read, no lock. Keep them pure CPU work on data already in hand. A Pydantic model_validate is exactly that.

One consequence worth knowing: validation happens on the event loop's thread, so a very large document validated on thousands of rows is CPU time the loop cannot use for I/O. For a bulk export of many rows where the document is not needed, select the columns you want instead of whole entities, or use model_construct for a fast path:

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return self.model.model_construct(**value)      # no validation, no coercion

model_construct skips validation and coercion, so nested models stay dicts and datetime fields stay strings. It is right only when the data is known-good and the model is flat. The default should be model_validate.

Streaming. Because the processor runs per row, streaming a large table with yield_per validates incrementally rather than all at once, which keeps memory flat — the behaviour described in streaming large result sets with yield_per in SQLAlchemy applies unchanged.

Relationship to other TypeDecorator uses. This is the same mechanism as an encrypted column, and the same caveats about caching and cache_ok apply, as in writing a TypeDecorator for encrypted columns. Alembic also needs to know how to render the type in a migration, which is rendering custom types in Alembic autogenerate — without it, autogenerate emits PydanticJSONB() with no import and the migration fails to run.

Mutation Tracking, Defaults and Querying Inside the Document

Three practical problems come up immediately.

In-place mutation is not detected. Exactly as with a plain JSON or ARRAY column, changing the model in place leaves the attribute looking unchanged:

order.shipping.speed = "express"    # not detected: no UPDATE is emitted
await session.commit()

The explicit fix is a reassignment, which Pydantic makes tidy with model_copy:

order.shipping = order.shipping.model_copy(update={"speed": "express"})
await session.commit()

This is the option to prefer. It is visible at the call site, it works with frozen models, and it avoids wrapping every document. Making the Pydantic model frozen=True turns the silent failure above into an error, which is worth doing:

from pydantic import BaseModel, ConfigDict


class Shipping(BaseModel):
    model_config = ConfigDict(frozen=True)
    # ... fields as above

The alternative is flag_modified, which tells the session the attribute changed:

from sqlalchemy.orm.attributes import flag_modified

order.shipping.speed = "express"
flag_modified(order, "shipping")

Correct, and easy to forget. MutableDict.as_mutable() does not help here, because the Python value is a model rather than a dict.

Defaults. A mutable default on a column is a trap — default=Shipping(carrier="dhl") shares one instance across every new row. Use a callable:

    shipping: Mapped[Shipping] = mapped_column(
        PydanticJSONB(Shipping), default=lambda: Shipping(carrier="dhl")
    )

With a frozen model the shared instance is harmless, but the callable form is right either way. A server_default is also worth having so rows inserted outside the ORM get a valid document:

from sqlalchemy import text

    shipping: Mapped[Shipping] = mapped_column(
        PydanticJSONB(Shipping),
        default=lambda: Shipping(carrier="dhl"),
        server_default=text("""'{"carrier": "dhl", "speed": "standard"}'::jsonb"""),
    )

Querying inside the document. The type decorator changes nothing about SQL. The column is JSONB, so every operator still works, and the indexing rules in querying and indexing JSONB columns in SQLAlchemy apply:

from sqlalchemy import select

# Containment, index-assisted by a GIN index on the column.
stmt = select(Order).where(Order.shipping.contains({"speed": "express"}))

# Text extraction for an equality filter.
stmt = select(Order).where(Order.shipping["carrier"].astext == "ups")

One wrinkle: the bind processor also runs on values used in comparisons, so comparing against a partial dict can fail validation. Order.shipping.contains({"speed": "express"}) goes through the JSONB comparator rather than the decorator in current SQLAlchemy, but the safe habit for a whole-column comparison is to build a full model, or to cast the column:

from sqlalchemy import cast
from sqlalchemy.dialects.postgresql import JSONB

stmt = select(Order).where(
    cast(Order.shipping, JSONB).contains({"speed": "express"})
)

A hybrid property is the tidiest way to give a frequently filtered field a name in both Python and SQL, which is the technique in writing hybrid properties that work in Python and SQL.

Schema Evolution and Enforcement Outside the ORM

A JSONB document's schema changes without a migration, which is the point — and the reason it needs a plan. Two mechanisms cover it.

Version the document. Give the model a version field and an upgrade path, so every historical row still loads:

from typing import Any, Literal

from pydantic import BaseModel, model_validator


class Shipping(BaseModel):
    version: Literal[2] = 2
    carrier: str
    speed: str = "standard"
    signature_required: bool = False

    @model_validator(mode="before")
    @classmethod
    def upgrade(cls, data: Any) -> Any:
        if isinstance(data, dict) and data.get("version", 1) == 1:
            data = dict(data)
            data["signature_required"] = data.pop("needs_signature", False)
            data["version"] = 2
        return data

Rows are rewritten to version 2 whenever they are next saved. Whether that is enough depends on how often rows are written; when the upgrade path must eventually be deleted, finish the job with a backfill in batches, using the batching approach in processing large tables in batches with partitions.

The rules that make this sustainable are the ones from any forward-compatible format. New fields get defaults, so old documents validate. Removed fields are ignored rather than rejected — Pydantic's default behaviour, and a reason not to set extra="forbid" on stored documents even though it is tempting. Renames go through an upgrade step, never a bare rename. And a field's type never changes in place: add a new field and retire the old one, exactly as you would for a column in dropping a column safely during a rolling deploy.

Defend against non-ORM writers. The type decorator only runs in Python processes that use these models. A psql session, an ETL job, or a different service writing the same table bypasses it entirely. Where that matters, add a CHECK constraint for the invariants the application cannot survive without:

from sqlalchemy import CheckConstraint


class Order(Base):
    __tablename__ = "orders"
    __table_args__ = (
        CheckConstraint(
            "shipping ? 'carrier' AND jsonb_typeof(shipping -> 'carrier') = 'string'",
            name="ck_orders_shipping_carrier",
        ),
    )

A CHECK cannot express a whole schema without becoming unmaintainable, so pick the two or three keys whose absence would break a code path, and let Pydantic handle the rest. Adding the constraint to an existing table needs the NOT VALID then VALIDATE sequence, because a plain ADD CONSTRAINT scans and locks the table — the pattern in adding a NOT NULL column without locking in Postgres.

A last consideration: keep genuinely relational fields out of the document. A field that is filtered on every request, joined against, or constrained by a foreign key belongs in a column. JSONB is for the parts of a record whose shape varies or whose schema is owned by something other than the database.

Resolving Warnings, Errors & Common Mistakes

Two notes on the table's fifth and sixth rows, which are the ones that bite in production rather than in development.

cache_ok. Omitting it produces a warning, not an error, and the application works — slowly, because SQLAlchemy's compiled-statement cache is bypassed for every statement touching the column. On a busy endpoint that is a large, silent regression. Set it, and keep the type's behaviour dependent only on its constructor arguments so that staying cacheable remains true.

Alembic rendering. env.py needs to know how to write the type into a migration script:

def render_item(type_, obj, autogen_context):
    if type_ == "type" and obj.__class__.__name__ == "PydanticJSONB":
        autogen_context.imports.add("import shop.types")
        autogen_context.imports.add("import shop.models")
        return f"shop.types.PydanticJSONB({obj.model.__module__}.{obj.model.__name__})"
    return False


context.configure(..., render_item=render_item)

There is a subtlety here worth stating plainly: a migration that references the current Pydantic model is not reproducible, because the model will change. For DDL purposes the type is just JSONB, so the simplest correct answer is often to render it as postgresql.JSONB(astext_type=sa.Text()) and keep the Python model out of migration history entirely. That keeps old migrations runnable after the model has moved on — the same reasoning that keeps enum values and imports out of migration scripts in reviewing autogenerated migrations before applying them.

Frequently Asked Questions

Related

• Querying Postgres JSONB, Arrays and Full-Text Search — The parent guide: operators and indexes for JSONB.
• Querying and indexing JSONB columns in SQLAlchemy — Filtering inside the document, and GIN indexes.
• Writing a TypeDecorator for encrypted columns — The same mechanism for a different purpose.
• Rendering custom types in Alembic autogenerate — Making migrations reference the type correctly.