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.14+ and Pydantic 2.11+. Tested with 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()

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.

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.

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.

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]

Rules and gotchas

• Every model inside the column should inherit EmbeddedPydanticModel, not pydantic.BaseModel. A plain BaseModel 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 plain models are fine only if they're never changed in place, for example frozen ones (model_config = ConfigDict(frozen=True)).
• Values are validated every time a row is loaded, against the current model. When you change a model, existing rows must still validate: give new fields a default (or update the stored rows), and handle renamed or removed fields, for example with a model_validator(mode="before") or a hand-written data migration.
• 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") 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 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.
• 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.

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).
• 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 2.0 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