sqlalchemy-pydantic-json

Store Pydantic models in SQLAlchemy JSON columns, and just change them in place: every change, however deeply nested, is saved when you commit.

No flag_modified() calls, no event listeners in your code, and full type-checker support (mypy, pyright and ty).

Why

A JSON column is a convenient place for structured data that doesn't deserve its own tables: settings, preferences, metadata. With plain SQLAlchemy you get dicts and lists back, and changing them in place isn't noticed: user.settings["theme"] = "dark" is silently lost unless you also call flag_modified(user, "settings"). SQLAlchemy's MutableDict helps for one level, but not for nested structures, and not for Pydantic models.

This package gives you real Pydantic models in the column (validation, defaults, types, autocompletion) and tracks every change inside them: fields, lists, dicts, sets and nested models, however deep.

Installation

pip install sqlalchemy-pydantic-json
# or
uv add sqlalchemy-pydantic-json

Requires Python 3.11+, SQLAlchemy 2.0.44+ and Pydantic 2.12+. Tested with SQLAlchemy 2.0 and 2.1, on SQLite, PostgreSQL and MariaDB, with both Session and AsyncSession.

Using Alembic? Then also do the one-time Alembic setup below. Without it, autogenerated migrations fail.

Quick start

Use EmbeddedPydanticModel as the base class for the column's model and for every model inside it, and declare the column with Model.column():

from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

from sqlalchemy_pydantic_json import EmbeddedPydanticModel


class Visit(EmbeddedPydanticModel):
    page: str = "/"


class Address(EmbeddedPydanticModel):
    city: str = "Helsinki"
    lines: list[str] = []


class Settings(EmbeddedPydanticModel):
    theme: str = "light"
    tags: set[str] = set()
    address: Address = Address()
    history: list[Visit] = []


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    settings: Mapped[Settings] = mapped_column(Settings.column(), default=Settings)
    extra: Mapped[Settings | None] = mapped_column(Settings.column())  # nullable


engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(User(id=1))
    session.commit()

    user = session.get(User, 1)
    user.settings.theme = "dark"
    user.settings.tags.add("admin")
    user.settings.address.lines.append("Mannerheimintie 1")
    user.settings.history.append(Visit(page="/home"))
    user.settings.history[0].page = "/start"
    assert user in session.dirty  # every change above marks the row as changed
    session.commit()

with Session(engine) as session:
    user = session.get(User, 1)
    assert user.settings.address.lines == ["Mannerheimintie 1"]
    assert user.settings.history[0].page == "/start"

You can also assign a whole model, or a dict (it's validated into the model), or None for a nullable column:

with Session(engine) as session:
    user = session.get(User, 1)
    user.settings = Settings(theme="blue")
    user.extra = {"theme": "green"}
    assert isinstance(user.extra, Settings)
    user.extra = None  # stored as SQL NULL
    session.commit()

What's tracked

Any of these changes marks the row as changed, at any depth:

• assigning or deleting a field (settings.address.city = "Oulu", del settings.theme)
• lists, dicts and sets: every method that changes them (append(), items[0] = ..., pop(), sort(), update(), add(), ...)
• lists, dicts and models inside tuples, also named tuples
• defaultdict, including the default that reading a missing key inserts; OrderedDict, including move_to_end(); and Counter
• the list or model in a root model
• extra values of a model with extra="allow"
• assigning a whole model, a dict or None to the column

Not tracked (see rules and gotchas):

• changes inside a plain pydantic.BaseModel submodel: use EmbeddedPydanticModel for every model
• changes inside a dataclass (a standard-library or a Pydantic one)
• changes inside a deque
• bulk and Core statements, such as session.execute(update(User).values(...))

PostgreSQL: JSON or JSONB

Model.column() uses SQLAlchemy's generic JSON type, which works on every database. On PostgreSQL that creates a json column. For jsonb (binary, indexable, more operators), pass json_type:

from sqlalchemy import JSON
from sqlalchemy.dialects.postgresql import JSONB

# always JSONB (PostgreSQL only)
settings: Mapped[Settings] = mapped_column(Settings.column(json_type=JSONB), default=Settings)

# JSONB on PostgreSQL, JSON elsewhere (e.g. SQLite in tests)
settings: Mapped[Settings] = mapped_column(
    Settings.column(
        json_type=JSON(none_as_null=True).with_variant(JSONB(none_as_null=True), "postgresql")
    ),
    default=Settings,
)

A type class such as JSONB automatically gets none_as_null=True, so that None is stored as SQL NULL. A type instance is used as is, so pass none_as_null=True yourself, as above.

On MySQL the generic JSON type maps to its native JSON type; on MariaDB, where JSON is the server's own alias for LONGTEXT with a validity check, it maps to that.

Querying inside the JSON

The column keeps SQLAlchemy's JSON operators, so you can filter on values inside the model:

with Session(engine) as session:
    blue = session.scalars(select(User).where(User.settings["theme"].as_string() == "blue")).all()
    in_helsinki = session.scalars(
        select(User).where(User.settings[("address", "city")].as_string() == "Helsinki")
    ).all()
    assert [u.id for u in blue] == [1]

See SQLAlchemy's JSON type documentation for the operators, and what each database supports.

Lists and unions as the column (root models)

For a column that holds a list, or one of several models, use EmbeddedPydanticRootModel, this package's version of Pydantic's RootModel. As with any root model, the list or model itself is its root attribute (customer.addresses.root.append(...)), and changes there are tracked like changes to any other field:

from typing import Annotated, Literal

from pydantic import Field

from sqlalchemy_pydantic_json import EmbeddedPydanticRootModel


class Card(EmbeddedPydanticModel):
    kind: Literal["card"] = "card"
    last_digits: str = ""


class Invoice(EmbeddedPydanticModel):
    kind: Literal["invoice"] = "invoice"
    emails: list[str] = []


AnyPayment = Annotated[Card | Invoice, Field(discriminator="kind")]


class Payment(EmbeddedPydanticRootModel[AnyPayment]):
    pass


class Addresses(EmbeddedPydanticRootModel[list[Address]]):
    pass


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    payment: Mapped[Payment] = mapped_column(Payment.column(), default=lambda: Payment(Card()))
    addresses: Mapped[Addresses] = mapped_column(Addresses.column(), default=lambda: Addresses([]))


Base.metadata.create_all(engine)

with Session(engine) as session:
    customer = Customer(id=1)
    session.add(customer)
    session.commit()

    customer.addresses.root.append(Address(city="Tampere"))
    invoice = Invoice(emails=["billing@example.com"])
    customer.payment = Payment(invoice)  # or `= invoice`: it's validated into a Payment
    invoice.emails.append("finance@example.com")  # tracked: it's the value in the column
    assert customer in session.dirty
    session.commit()

Generic models (class Box(EmbeddedPydanticModel, Generic[T])) work too, and so do models with extra="allow": their extra values are stored and tracked like fields.

Aliases (e.g. camelCase)

Pydantic aliases decide the key names in the stored JSON. For camelCase, make your own base class with an alias generator, and use it for all of your models:

from pydantic import ConfigDict, Field
from pydantic.alias_generators import to_camel


class CamelModel(EmbeddedPydanticModel):
    model_config = ConfigDict(alias_generator=to_camel, validate_by_name=True)


class Profile(CamelModel):
    display_name: str = "anon"  # stored as "displayName"
    tax_id: str | None = Field(default=None, alias="TIN")  # an explicit alias wins: "TIN"


class Member(Base):
    __tablename__ = "members"

    id: Mapped[int] = mapped_column(primary_key=True)
    profile: Mapped[Profile] = mapped_column(Profile.column(), default=Profile)


Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Member(id=1, profile=Profile(display_name="Jocke", TIN="123")))
    session.commit()  # stored as {"displayName": "Jocke", "TIN": "123"}

    query = select(Member.id).where(Member.profile["displayName"].as_string() == "Jocke")
    assert session.scalars(query).all() == [1]

• The JSON is stored with the aliases, as by Pydantic's model_dump(by_alias=True). Loading accepts both the aliases and the field names, so rows stored before you added an alias still load. They're stored with the aliases the next time they're saved.
• Queries into the JSON use the stored names: Member.profile["displayName"].
• validate_by_name=True lets your Python code use field names. Type checkers know the field names of generated aliases (display_name=), but only the alias of an explicit Field(alias="TIN") (TIN=), so write it that way. Also pass Field(default=...) as a keyword: type checkers treat a positional default as a required field.
• A field must be loadable from the name it's stored under. If its serialization_alias differs from its validation_alias, defining the model raises a TypeError (include the stored name with AliasChoices if you need both).

Default values

These all work:

# a new model for each row (recommended)
settings: Mapped[Settings] = mapped_column(Settings.column(), default=Settings)

# a dict, validated into a new model for each row
settings: Mapped[Settings] = mapped_column(Settings.column(), default={"theme": "dark"})

# a default in the database; the model's own field defaults fill in the rest when loaded
settings: Mapped[Settings] = mapped_column(Settings.column(), server_default=text("'{}'"))

Don't use a model instance as the default (default=Settings()): SQLAlchemy then puts that same object into every new row, so changing one row's settings changes all of them.

With dataclass-style mapping (MappedAsDataclass), use default_factory=Settings.

Alembic setup

Alembic's autogenerate can't write the column type into a migration by itself: it would write sqlalchemy_pydantic_json._model.PydanticJSON(...), which fails when the migration runs. Tell it to write the plain JSON type instead. In your env.py, pass render_item to both context.configure() calls (offline and online):

from sqlalchemy_pydantic_json.alembic import make_render_item

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

If you already have a render_item function of your own, wrap it:

context.configure(..., render_item=make_render_item(wrap=my_render_item))

Migrations then contain sa.JSON(none_as_null=True) (or postgresql.JSONB(...), or the variant), and depend on neither this package nor your models, so they keep working as your models change.

Changing the model doesn't change the database schema, so Alembic has nothing to generate for it. Existing rows must still validate against the new model, though: after adding a required field or renaming one, say, either make the model accept the old data, or update the stored JSON yourself (for example in a hand-written data migration). Changing the column's type between JSON and JSONB is detected like any other type change.

Using with SQLModel

Declare the column with sa_column:

from sqlalchemy import Column
from sqlmodel import Field, SQLModel, col, select
from sqlmodel import Session as SQLModelSession


class Player(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    settings: Settings = Field(
        default_factory=Settings,
        sa_column=Column(Settings.column(), nullable=False),
    )


SQLModel.metadata.create_all(engine)

with SQLModelSession(engine) as session:
    session.add(Player(id=1))
    session.commit()

    player = session.get(Player, 1)
    player.settings.tags.add("captain")  # tracked, as with SQLAlchemy models
    assert player in session.dirty
    session.commit()

    query = select(Player.id).where(col(Player.settings)["theme"].as_string() == "light")
    assert session.exec(query).all() == [1]

Using with FastAPI

The models work as FastAPI request and response models, like any Pydantic model. You can assign a request body straight to a row (user.settings = settings), and return a row's value as the response (return user.settings).

Rules and gotchas

• Every model inside the column should inherit EmbeddedPydanticModel, not pydantic.BaseModel, and not be a dataclass. A plain BaseModel or a dataclass still loads and saves correctly, and replacing it as a whole is tracked, but changes inside it aren't: they're lost unless something else in the row changes too. So they're fine only if they're never changed in place, for example frozen ones (model_config = ConfigDict(frozen=True)) with nothing changeable in them: a list in a frozen model can still be appended to. (In a frozen EmbeddedPydanticModel, that's tracked.)
• Changes inside a deque aren't tracked: neither append() and the like, nor changes to the lists or models in it. Assign a new deque (settings.queue = deque(...)) to store a change, or use a list: JSON has no deque, so it's stored as a list anyway.
• Values are validated every time a row is loaded, against the current model. When you change a model, existing rows must still validate:
  • give a new field a default (or update the stored rows);
  • to rename a field, keep loading its old name too with new_name: str = Field(validation_alias=AliasChoices("new_name", "old_name")): a row is stored under the new name the next time it's saved;
  • for anything else, use a model_validator(mode="before") or a hand-written data migration.
• Computed fields and excluded fields aren't stored. A @computed_field is calculated again when the row is loaded, so you can't query it inside the JSON. A field with Field(exclude=True) isn't saved at all, and loads as its default.
• Bulk and Core statements bypass tracking, as with any SQLAlchemy attribute: session.execute(update(User).values(...)) writes what you give it, and doesn't know about in-place changes.
• Values you keep across an expiring commit are no longer tracked. With the default expire_on_commit=True, commit() expires the row; the next access loads a fresh model. Changing the old model you kept a reference to does nothing (and doesn't raise). Read the value from the row again after committing.
• Shallow copies share nested models, as in Pydantic: changing a nested model in a model_copy() or copy.copy() also changes it in the original. Use model_copy(deep=True) or copy.deepcopy() for an independent copy. Copies (and pickled models) aren't attached to any row until you assign them.
• Thread safety is the same as for SQLAlchemy sessions: don't share one between threads.

How it works

• PydanticJSON is a SQLAlchemy TypeDecorator over JSON: it validates the model on load and dumps it with model_dump(mode="json", by_alias=True, exclude_computed_fields=True) on save. On its own it doesn't track anything.
• EmbeddedPydanticModel combines Pydantic's BaseModel with SQLAlchemy's Mutable, and Model.column() is Model.as_mutable(PydanticJSON(Model)).
• Whenever a field (or an extra value, with extra="allow") is set, lists, dicts and sets are wrapped in tracked versions of SQLAlchemy's MutableList, MutableDict and MutableSet, and nested models are linked to their parent. A tuple never changes, so the values inside it are linked to the tuple's parent instead. A deque is left as it is.
• A defaultdict, OrderedDict or Counter becomes a tracked subclass of its own type, so it keeps its methods. The defaultdict one builds on the tracked dict. The other two hook their own methods: MutableDict changes a dict with dict's own methods, which would skip an OrderedDict's bookkeeping of the order, and its update() would replace a Counter's counts instead of adding to them.
• Each model or container keeps weak references to all of its parents. A change is passed up from parent to parent until it reaches the model in the column, which marks the row as changed. A parent that no longer holds the value (after a pop() or reassignment, say) is skipped and forgotten, so values can be moved around and shared freely.
• Each link also remembers where the parent holds the value (a list index, dict key or field name), so checking it is a single lookup, even in long lists. Only a value that has moved is searched for, once.

Alternatives

• sqlalchemy-json: nested change tracking for plain dicts and lists, without Pydantic models.
• SQLAlchemy-Nested-Mutable: nested tracking including Pydantic models, but for Pydantic v1 only.
• SQLModel: Pydantic and SQLAlchemy in one model class, but no built-in change tracking for Pydantic models in JSON columns. This package adds it (see above).
• The TypeDecorator recipe that gets passed around: it converts models to and from JSON, but doesn't notice in-place changes, so you still call flag_modified() yourself.
• activemodel: an ActiveRecord-style framework on top of SQLModel. Its PydanticJSONMixin also tracks changes in Pydantic models in JSON columns, by comparing snapshots of the JSON when the session commits. It requires SQLModel, and a change isn't visible to flushes (including autoflush before a query) until then.

This package needs only SQLAlchemy and Pydantic. It works with SQLAlchemy's declarative models and with SQLModel, and notices every change the moment it's made.

Contributing

See CONTRIBUTING.md. Changes are listed in the changelog.

License

MIT