Metadata-Version: 2.4
Name: sqlalchemy-pydantic-json
Version: 0.2.1
Summary: Pydantic models in SQLAlchemy JSON columns, with automatic mutation tracking
Keywords: sqlalchemy,pydantic,json,jsonb,mutable,mutation-tracking,orm
Author: Joakim Nordling
Author-email: Joakim Nordling <joakim.nordling@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Pydantic :: 2
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Typing :: Typed
Requires-Dist: sqlalchemy>=2.0.44
Requires-Dist: pydantic>=2.12
Requires-Python: >=3.11
Project-URL: Homepage, https://github.com/joakimnordling/sqlalchemy-pydantic-json
Project-URL: Documentation, https://github.com/joakimnordling/sqlalchemy-pydantic-json#readme
Project-URL: Changelog, https://github.com/joakimnordling/sqlalchemy-pydantic-json/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/joakimnordling/sqlalchemy-pydantic-json/issues
Description-Content-Type: text/markdown

# sqlalchemy-pydantic-json

[![PyPI](https://img.shields.io/pypi/v/sqlalchemy-pydantic-json)](https://pypi.org/project/sqlalchemy-pydantic-json/)
[![Python versions](https://img.shields.io/pypi/pyversions/sqlalchemy-pydantic-json)](https://pypi.org/project/sqlalchemy-pydantic-json/)
[![CI](https://github.com/joakimnordling/sqlalchemy-pydantic-json/actions/workflows/ci.yml/badge.svg)](https://github.com/joakimnordling/sqlalchemy-pydantic-json/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/pypi/l/sqlalchemy-pydantic-json)](https://github.com/joakimnordling/sqlalchemy-pydantic-json/blob/main/LICENSE)
[![Coverage: 100%](https://img.shields.io/badge/coverage-100%25-brightgreen)](https://github.com/joakimnordling/sqlalchemy-pydantic-json/actions/workflows/ci.yml)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

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

```bash
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](#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()`:

```python
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:

```python
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](#lists-and-unions-as-the-column-root-models)
- 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](#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`:

<!-- readme-test: skip -->
```python
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, and
select them:

```python
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]

    # values selected from inside the JSON are the raw JSON, not validated by Pydantic
    theme, tags, address = session.execute(
        select(User.settings["theme"], User.settings["tags"], User.settings["address"])
    ).one()
    assert theme == "blue"
    assert tags == []  # a list, not a set
    assert address == {"city": "Helsinki", "lines": []}  # a dict, not an Address
    assert Address.model_validate(address) == Address()
```

A value selected from inside the JSON is what's stored there. Pydantic doesn't validate it, so you
don't get your model's types: a submodel comes back as a dict, a set as a list, and a `datetime`,
`UUID`, `Decimal` or enum as the string or number it's stored as. The keys are the stored names,
that is, the aliases if your models have any (see [aliases](#aliases-eg-camelcase)). Validate the
value yourself if you need the model, as with `Address.model_validate()` above. Selecting only
the part you need also skips loading and validating the whole model, which can help with large
documents.

Compare values with a typed accessor such as `.as_string()` or `.as_integer()`, as above. Comparing
the JSON value directly (`User.settings["theme"] == "blue"`) works differently on each database,
as with any JSON column. With `JSONB`, its own operators work too, e.g.
`User.settings.contains({"theme": "blue"})` or `User.settings.has_key("theme")`.

See SQLAlchemy's [JSON type documentation](https://docs.sqlalchemy.org/en/20/core/type_basics.html#sqlalchemy.types.JSON)
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`](https://docs.pydantic.dev/latest/concepts/models/#rootmodel-and-custom-root-types).
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:

```python
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:

```python
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:

<!-- readme-test: skip -->
```python
# 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):

<!-- readme-test: skip -->
```python
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:

<!-- readme-test: skip -->
```python
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`:

<!-- readme-test: needs sqlmodel -->
```python
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](https://github.com/edelooff/sqlalchemy-json): nested change tracking for plain
  dicts and lists, without Pydantic models.
- [SQLAlchemy-Nested-Mutable](https://github.com/wonderbeyond/sqlalchemy-nested-mutable): nested
  tracking including Pydantic models, but for Pydantic v1 only.
- [SQLModel](https://sqlmodel.tiangolo.com/): 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](#using-with-sqlmodel)).
- [The `TypeDecorator` recipe](https://gist.github.com/imankulov/4051b7805ad737ace7d8de3d3f934d6b)
  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](https://github.com/iloveitaly/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](https://github.com/joakimnordling/sqlalchemy-pydantic-json/blob/main/CONTRIBUTING.md).
Changes are listed in the [changelog](https://github.com/joakimnordling/sqlalchemy-pydantic-json/blob/main/CHANGELOG.md).

## License

[MIT](https://github.com/joakimnordling/sqlalchemy-pydantic-json/blob/main/LICENSE)
