Metadata-Version: 2.4
Name: sqlalchemy-pydantic-json
Version: 0.0.1a1
Summary: Pydantic v2 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 :: 3 - Alpha
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.14
Requires-Dist: pydantic>=2.11
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

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

## 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.

## Querying inside the JSON

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

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

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.

## 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.

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

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

## 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](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)).
- [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 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](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)
