Mutations and `MutableDict`: the gotcha that loses your changes silently

Capsule overview

In module 1 you mastered JSONB in pure SQL. In the previous capsule you learned to read and filter it from SQLAlchemy. Now comes the moment where the ORM betrays you: what seems trivial — post.metadata_["views"] = 1234 and then await session.commit() — has a gotcha that silently loses your changes. SQLAlchemy doesn't detect the mutation. The session commits. The query doesn't generate an UPDATE. The database still has the old value. And your green test doesn't warn you.

This capsule teaches you why it happens, the two ways to fix it (MutableDict.as_mutable(JSONB) for in-place mutations and func.jsonb_set for partial updates), and when each one is best. More importantly: you learn to recognize the symptom. If you see "I saved it but it wasn't persisted" in JSONB again, this is the first suspect.

By the end you'll be able to write code that mutates JSONB and actually persists it, you'll be able to explain to a colleague why their silent update fails, and you'll be able to choose between the two fixes depending on the case.

Mental model: SQLAlchemy's change tracking system

To understand the bug you have to understand how SQLAlchemy detects changes. The simplified rule is:

SQLAlchemy detects changes when you reassign an attribute (obj.x = new_value). It does NOT detect changes when you mutate an existing value (obj.x.append(...), obj.x["k"] = v).

Every object in a session has an internal "state." When you reassign obj.x = y, SQLAlchemy compares y with the previous value and marks the attribute as "dirty" (modified). On commit(), it generates an UPDATE ... SET x = y for each dirty attribute.

When you mutate in-place (obj.x["k"] = v), the referenced object is the same one. SQLAlchemy doesn't find out. The internal state still says "this attribute didn't change." On commit(), it doesn't generate an UPDATE.

For immutable types this doesn't matter: you can't mutate an int or a str. For mutable types (dict, list, set), it does matter. JSONB is loaded in Python as a mutable dict or list. That's why the bug appears.

                         ┌──────────────────────────────────────┐
                         │  SQLAlchemy session                  │
                         │                                      │
  reassignment   ───→    │  obj.attr = new                      │
  (X = Y)               │  → marks it "dirty"                  │
                         │  → UPDATE on commit()                │
                         │                                      │
                         │  ─────────────────                   │
                         │                                      │
  in-place       ───→    │  obj.attr["k"] = v                   │
  mutation               │  → same object referenced            │
                         │  → does NOT mark it "dirty"          │
                         │  → NO UPDATE on commit() ❌          │
                         │                                      │
                         └──────────────────────────────────────┘

Memorize that box. It's the basis of the whole problem.

The bug in code

Let's reproduce it end-to-end. First the model, with nothing special:

# bug_demo.py
import asyncio
from typing import Any

from sqlalchemy import BigInteger, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata", JSONB, nullable=False, default=dict
    )


async def main() -> None:
    engine = create_async_engine(
        "postgresql+asyncpg://postgres:postgres@localhost:5432/demo",
        echo=True,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    Session = async_sessionmaker(engine, expire_on_commit=False)

    # Create a post
    async with Session() as session:
        post = Post(
            title="Demo",
            metadata_={"views": 0, "tags": ["intro"]},
        )
        session.add(post)
        await session.commit()
        post_id = post.id

    # Try to mutate in-place
    async with Session() as session:
        post = (await session.execute(select(Post).where(Post.id == post_id))).scalar_one()
        print("BEFORE:", post.metadata_)

        # In-place mutation — this is what does NOT get persisted
        post.metadata_["views"] = 9999
        post.metadata_["tags"].append("updated")

        await session.commit()  # silently: it doesn't generate an UPDATE for 'metadata'

    # Verify
    async with Session() as session:
        post = (await session.execute(select(Post).where(Post.id == post_id))).scalar_one()
        print("AFTER:", post.metadata_)

    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

Output (summary):

BEFORE: {'views': 0, 'tags': ['intro']}

-- In the log with echo=True NO UPDATE appears for posts.metadata
-- (only the initial SELECT and nothing else)

AFTER: {'views': 0, 'tags': ['intro']}

The change was lost. SQLAlchemy didn't mark metadata_ as dirty. It didn't generate an UPDATE. The database is still as it was.

Why this bug is so expensive:

• It's silent. There's no error, no warning, no log. The session commits and everything looks fine.
• It passes the happy tests. If your test is "I modify and read back in the same session," the in-memory dict reflects the change (because you did modify it). The test passes. The bug only shows up when you reopen a new session or query from another process.
• It's the natural pattern in Python. obj.metadata["x"] = v is what any dev would write without thinking about it. There's no intuitive reason to suspect it fails.

Solution 1: MutableDict.as_mutable(JSONB) — detecting in-place mutations

SQLAlchemy offers a wrapper in sqlalchemy.ext.mutable that does track mutations. You use it like this:

# models.py, fixed
from typing import Any

from sqlalchemy import BigInteger
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",
        MutableDict.as_mutable(JSONB),   # ← the key part
        nullable=False,
        default=dict,
    )

MutableDict.as_mutable(JSONB) wraps the JSONB type with a proxy that intercepts __setitem__, __delitem__, update, pop, etc. Every time you mutate the dict, the proxy notifies the session that the attribute changed.

Now the buggy code works:

async with Session() as session:
    post = (await session.execute(select(Post).where(Post.id == post_id))).scalar_one()
    post.metadata_["views"] = 9999
    await session.commit()
    # → generates: UPDATE posts SET metadata = $1 WHERE id = $2
    #              with $1 = '{"views": 9999, "tags": ["intro"]}'

An important limitation of MutableDict: it isn't recursive

MutableDict tracks mutations of the root dict, but it doesn't track mutations of nested objects. If your JSONB has this structure:

post.metadata_ = {
    "seo": {"title": "Hello", "tags": ["a", "b"]}
}

And you do:

post.metadata_["seo"]["title"] = "New"   # ❌ NOT detected
post.metadata_["seo"]["tags"].append("c")  # ❌ NOT detected

That isn't detected either. MutableDict only finds out about changes to the first-level dict. The inner dict {"title": "Hello", ...} is a normal dict, not a MutableDict.

How to work with this:

Option A — reassignment at the root level:

post.metadata_["seo"] = {**post.metadata_["seo"], "title": "New"}
# Reassigning the "seo" key of the root dict does trigger MutableDict.

Option B — explicit flag_modified:

from sqlalchemy.orm.attributes import flag_modified

post.metadata_["seo"]["title"] = "New"
flag_modified(post, "metadata_")
# Manually tells the session: "this attribute changed, mark it dirty."

Option C — recursive MutableDict.coerce (advanced version): sqlalchemy.ext.mutable.MutableDict with a custom coerce exists to make it recursive, but the standard and maintainable pattern is A or B, or jumping to solution 2 (func.jsonb_set).

The trade-off of MutableDict

Pros:

• Natural Python syntax: obj.metadata_["k"] = v.
• No changes to the code pattern you already write.
• Good for flat JSONs or ones with a single level of nesting.

Cons:

• It doesn't detect deep mutations without reassignment or flag_modified.
• It rewrites the whole column in the UPDATE: UPDATE posts SET metadata = '{...the entire JSON...}'. If the JSONB weighs 500 KB and you only changed one field, you're still sending 500 KB on every commit.
• Row-level locking during the UPDATE, which is problematic with high concurrency on the same record.

For small JSONs (<10 KB) and mutations from a single actor, MutableDict is fine. For large JSONs, high concurrency, or very targeted updates, solution 2 is better.

Solution 2: func.jsonb_set — a partial update without rewriting the whole column

jsonb_set is the PostgreSQL function you already know from module 1. It modifies a specific path inside a JSONB and returns the modified JSONB. From SQLAlchemy it's invoked with func.jsonb_set:

from sqlalchemy import func, update

# A targeted UPDATE without loading the object into memory
stmt = (
    update(Post)
    .where(Post.id == post_id)
    .values(
        metadata_=func.jsonb_set(
            Post.metadata_,
            "{views}",        # path as a PostgreSQL string '{key1,key2,...}'
            "9999",           # new value (JSON text)
        )
    )
)
await session.execute(stmt)
await session.commit()

Generated SQL:

UPDATE posts
SET metadata = jsonb_set(posts.metadata, '{views}', '9999')
WHERE posts.id = $1;

PostgreSQL does the merge atomically. You don't rewrite the whole column from Python — the engine modifies only the indicated path.

Building the path and the value safely

jsonb_set expects two things:

1. A path as a PostgreSQL text[]: '{key1,key2,key3}' or ARRAY['key1','key2','key3'].
2. A value as JSONB: a string that is valid JSON ('"hello"', '9999', '{"k":"v"}').

To avoid concatenating strings by hand, use helpers:

import json
from sqlalchemy import func, cast
from sqlalchemy.dialects.postgresql import JSONB

def jsonb_set_path(column, path: list[str], value):
    """
    Helper that wraps func.jsonb_set to avoid quoting errors.
    """
    pg_path = "{" + ",".join(path) + "}"
    return func.jsonb_set(column, pg_path, cast(json.dumps(value), JSONB))


# Usage:
stmt = (
    update(Post)
    .where(Post.id == post_id)
    .values(metadata_=jsonb_set_path(Post.metadata_, ["seo", "title"], "New title"))
)

json.dumps("New title") produces the string '"New title"', which is valid JSON for a string. json.dumps({"k": "v"}) produces '{"k": "v"}', which is valid JSON for an object. cast(..., JSONB) converts it to the correct type in PostgreSQL.

Creating keys that don't exist

By default jsonb_set only modifies if the path exists. If you want it created when it doesn't exist, pass create_missing=True (the fourth argument, True by default in SQL, but it's better to be explicit):

stmt = (
    update(Post)
    .where(Post.id == post_id)
    .values(
        metadata_=func.jsonb_set(
            Post.metadata_,
            "{new_field}",
            cast('"value"', JSONB),
            True,   # create_missing
        )
    )
)

Careful: if create_missing is True (the default) and the path is {a,b,c} but a doesn't exist, the chain of intermediate objects isn't created. jsonb_set only creates the last key of the path if the parents exist. To create deep structures from scratch, chain several jsonb_set calls or build the sub-object and merge:

# If you need to ensure metadata.seo exists before setting metadata.seo.title:
stmt = (
    update(Post)
    .where(Post.id == post_id)
    .values(
        metadata_=func.jsonb_set(
            func.coalesce(
                Post.metadata_,
                cast("{}", JSONB),
            ),
            "{seo,title}",
            cast('"New"', JSONB),
        )
    )
)
# If metadata.seo doesn't exist, this will fail silently by not creating "seo".
# To create "seo" first, chain two jsonb_set calls:
stmt2 = (
    update(Post)
    .where(Post.id == post_id)
    .values(
        metadata_=func.jsonb_set(
            func.jsonb_set(
                Post.metadata_, "{seo}", cast("{}", JSONB), True
            ),
            "{seo,title}",
            cast('"New"', JSONB),
        )
    )
)

It's verbose. If your nested update logic gets complicated, consider loading the object, using MutableDict with flag_modified, and accepting the cost of rewriting the column. There's a point where simple wins.

The trade-off of func.jsonb_set

Pros:

• An atomic update at the PostgreSQL level — only the path changes.
• It doesn't require loading the object into memory or reading the whole JSONB.
• Good for high concurrency: two transactions modifying different paths don't stomp on the whole JSON.
• Good for large JSONs where rewriting everything is expensive.

Cons:

• Verbose syntax compared to obj.metadata_["k"] = v.
• If the path doesn't exist, the intermediates aren't created automatically — you have to chain jsonb_set.
• You need to write a helper so you don't get the quoting of paths and values wrong.
• It doesn't update the object in memory — if you need the new value afterwards in the same session, refresh: await session.refresh(post).

Decision matrix: when MutableDict vs func.jsonb_set

Practical rule: start with MutableDict so you don't complicate the code. Switch to func.jsonb_set when a specific case shows up where it wins (concurrency, large JSONs, targeted updates in jobs).

Worked example: applying both solutions to a real case

The case: your Blog API needs a POST /posts/{id}/views endpoint that increments the view counter stored in metadata.views. And a PATCH /posts/{id}/seo endpoint that updates the metadata.seo sub-object with partial fields coming from the client.

# blog_api.py
import asyncio
import json
from typing import Any

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import BigInteger, cast, func, select, update
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",
        MutableDict.as_mutable(JSONB),
        nullable=False,
        default=dict,
    )


engine = create_async_engine(
    "postgresql+asyncpg://postgres:postgres@localhost:5432/demo"
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


app = FastAPI()


class SEOPatch(BaseModel):
    title: str | None = None
    description: str | None = None
    canonical: str | None = None


@app.post("/posts/{post_id}/views")
async def increment_views(post_id: int) -> dict[str, int]:
    """
    Uses func.jsonb_set: high concurrency on this endpoint (every visit
    triggers an update). An atomic update at the level of the 'views' path.
    It doesn't load the object into memory.
    """
    async with SessionLocal() as session:
        # Read the current value and add 1 at the SQL level.
        stmt = (
            update(Post)
            .where(Post.id == post_id)
            .values(
                metadata_=func.jsonb_set(
                    Post.metadata_,
                    "{views}",
                    cast(
                        # Coalesce to treat a missing 'views' as 0
                        func.to_jsonb(
                            cast(
                                func.coalesce(
                                    Post.metadata_["views"].astext, "0"
                                ),
                                BigInteger,
                            )
                            + 1
                        ),
                        JSONB,
                    ),
                    True,
                )
            )
            .returning(Post.metadata_["views"].astext.label("views"))
        )
        result = await session.execute(stmt)
        row = result.first()
        if row is None:
            raise HTTPException(404, "post not found")
        await session.commit()
        return {"views": int(row.views)}


@app.patch("/posts/{post_id}/seo")
async def patch_seo(post_id: int, patch: SEOPatch) -> dict[str, Any]:
    """
    Uses MutableDict + flag_modified: updating the seo sub-object
    with partial fields. Small JSON, the merge logic is more natural
    in Python.
    """
    from sqlalchemy.orm.attributes import flag_modified

    async with SessionLocal() as session:
        post = (
            await session.execute(select(Post).where(Post.id == post_id))
        ).scalar_one_or_none()
        if post is None:
            raise HTTPException(404, "post not found")

        seo_existing = post.metadata_.get("seo", {})
        # Merge the non-None fields of the patch
        patch_dict = patch.model_dump(exclude_none=True)
        merged = {**seo_existing, **patch_dict}

        # Reassignment at the root level: MutableDict detects this
        post.metadata_["seo"] = merged

        await session.commit()
        return post.metadata_["seo"]

Why each endpoint uses a different solution:

• /posts/{id}/views is high concurrency (potentially thousands per second on a viral post). func.jsonb_set avoids race conditions: the increment is atomic in PostgreSQL. If two requests arrive simultaneously, neither loses its +1.

• /posts/{id}/seo is low frequency (an editor changes a post's SEO when they edit it). The merge of partial fields is more readable in Python than as nested func.jsonb_set calls. MutableDict detects the reassignment of post.metadata_["seo"] = merged and persists it.

If you had written /posts/{id}/views with MutableDict (post.metadata_["views"] += 1), it would work until two concurrent requests read the same initial value and save the same +1, losing a view. The classic "lost update" bug.

If you had written /posts/{id}/seo with func.jsonb_set, you'd have to chain three calls (one per patch field) or do the merge in SQL. More verbose, same result.

Why does this matter in real work?

1. It's the #1 JSONB-with-an-ORM bug in production. SQLAlchemy community surveys confirm it year after year: the ticket "I save changes and they don't persist" is the most common one when using JSONB. Knowing the pattern before you smash your head into it saves you from debugging at night.

2. Tests pass, production breaks. The typical tests modify and verify in the same session. The in-memory dict does reflect the change (because you did modify it). The bug only shows up when you reopen a new session or query from another process. Green test, red prod. If your test doesn't include a commit and a new session, you don't detect this bug.

3. Meaningful code reviews. When a colleague does obj.json_field["k"] = v, you already know it has to have MutableDict or use func.jsonb_set. You flag it in the review and save them the "it doesn't save" ticket.

4. A concurrency decision. Knowing that MutableDict rewrites the whole column makes you think twice before using it in high-concurrency endpoints. func.jsonb_set is the tool when two transactions could stomp on each other.

Traps and common mistakes

Mistake 1 (conceptual): assuming the ORM detects any change

Symptom: you modify a dict, commit, it isn't persisted, and you don't understand why.

Why it happens: SQLAlchemy detects reassignments, not in-place mutations. For mutable types like dict and list, it doesn't find out about the changes without help.

How to detect it: run with echo=True and check whether the UPDATE appears in the log. If it doesn't appear, it wasn't generated.

How to fix it: MutableDict.as_mutable(JSONB) on the column or func.jsonb_set for the specific update.

Mistake 2 (conceptual): thinking MutableDict is recursive

Symptom: you declared MutableDict.as_mutable(JSONB), mutated a deep sub-object (obj.metadata_["seo"]["title"] = "X"), committed, and it isn't persisted.

Why it happens: MutableDict only tracks changes to the root dict. Nested dicts are normal dicts.

How to fix it: reassign at the root (obj.metadata_["seo"] = {**obj.metadata_["seo"], "title": "X"}) or use flag_modified(obj, "metadata_") after the deep mutation.

Mistake 3 (practical): forgetting expire_on_commit=False

Symptom: after the commit, you try to access post.metadata_ and get a closed-session error or a LazyLoadError.

Why it happens: by default, SQLAlchemy expires objects after the commit (it marks them as "stale," forcing a reload). If your session closed or is async without expire_on_commit=False, accessing the attribute triggers a reload that can fail.

Fix: async_sessionmaker(engine, expire_on_commit=False). It's what the official docs recommend for async. For sync sessions, Session(expire_on_commit=False) or configure it in the sessionmaker.

Mistake 4 (practical): func.jsonb_set with a path string and strange characters

Symptom: your key has a space or a special character ("open graph", "key,with,commas"), and func.jsonb_set(col, "{open graph}", ...) doesn't work the way you expect.

Why it happens: the '{a,b,c}' format is a PostgreSQL array literal. The commas separate elements. Spaces and special characters in keys require quoting.

Fix: avoid keys with spaces or special characters in JSONB (it's good style). If you can't, use the correct quoting:

# For a key with a space:
func.jsonb_set(Post.metadata_, '{"open graph"}', cast('"value"', JSONB))

Or build the path with SQLAlchemy's array(...):

from sqlalchemy import literal_column
func.jsonb_set(Post.metadata_, literal_column("ARRAY['open graph']"), cast('"value"', JSONB))

Mistake 5 (practical): refreshing the object after func.jsonb_set

Symptom: you run update(...).values(metadata_=func.jsonb_set(...)), commit, and the in-memory object still has the old data.

Why it happens: the update(...) executes at the SQL level without going through the session. The Python object doesn't find out about the change that happened in PostgreSQL.

Fix: after the commit, refresh the object:

await session.refresh(post)
print(post.metadata_)  # now it does reflect the change

Or use .returning(Post.metadata_) in the update to get the updated value in the same operation.

Mistake 6 (conceptual): tests that don't detect the bug

Symptom: green test, red prod. The test says it was saved. The database doesn't have the change.

Why it happens: the test modifies, commits, and verifies in the same session without reopening it. The in-memory dict does reflect the change. SQLAlchemy never generated an UPDATE but the test doesn't notice.

Fix: every test that verifies JSONB persistence must:

1. Modify and commit in one session.
2. Close the session.
3. Open a new session.
4. Read again and verify.

async def test_metadata_persists():
    async with SessionLocal() as s1:
        post = (await s1.execute(select(Post).where(Post.id == 1))).scalar_one()
        post.metadata_["views"] = 42
        await s1.commit()

    async with SessionLocal() as s2:  # a new session
        post = (await s2.execute(select(Post).where(Post.id == 1))).scalar_one()
        assert post.metadata_["views"] == 42  # this would fail without MutableDict

Exercises

Exercise 1: identify the bug

The following code is meant to increment views and add a tag. After running it, the changes aren't reflected when you query from another session. What's happening? Give two different fixes.

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata", JSONB, nullable=False, default=dict
    )

# ...
async with SessionLocal() as session:
    post = (await session.execute(select(Post).where(Post.id == 1))).scalar_one()
    post.metadata_["views"] += 1
    post.metadata_.setdefault("tags", []).append("popular")
    await session.commit()

from sqlalchemy.ext.mutable import MutableDict

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",
        MutableDict.as_mutable(JSONB),
        nullable=False,
        default=dict,
    )

tags = post.metadata_.get("tags", [])
post.metadata_["tags"] = [*tags, "popular"]

from sqlalchemy import func, cast, update
from sqlalchemy.dialects.postgresql import JSONB

stmt = (
    update(Post)
    .where(Post.id == 1)
    .values(
        metadata_=func.jsonb_set(
            func.jsonb_set(
                Post.metadata_,
                "{views}",
                cast(
                    func.to_jsonb(
                        cast(
                            func.coalesce(Post.metadata_["views"].astext, "0"),
                            BigInteger,
                        ) + 1
                    ),
                    JSONB,
                ),
                True,
            ),
            "{tags}",
            cast(
                func.coalesce(
                    Post.metadata_["tags"], cast("[]", JSONB)
                ) || cast('["popular"]', JSONB),
                JSONB,
            ),
            True,
        )
    )
)
await session.execute(stmt)
await session.commit()

Exercise 2: predict the SQL

Assume the model with MutableDict.as_mutable(JSONB). For each block of code, say whether SQLAlchemy generates an UPDATE on the commit and what SQL it produces.

# Case A
post.metadata_ = {"a": 1}
await session.commit()

# Case B
post.metadata_["a"] = 2
await session.commit()

# Case C
post.metadata_["seo"]["title"] = "New"
await session.commit()

# Case D
post.metadata_.update({"x": 10, "y": 20})
await session.commit()

Exercise 3: write a helper for func.jsonb_set

Write a Python function set_jsonb_path(column, path: list[str], value) that wraps func.jsonb_set correctly: it serializes the value with json.dumps, builds the path in PostgreSQL format, and does a cast to JSONB.

import json
from sqlalchemy import func, cast
from sqlalchemy.dialects.postgresql import JSONB


def set_jsonb_path(column, path: list[str], value, create_missing: bool = True):
    """
    An idiomatic Python equivalent of jsonb_set.

    Args:
        column: the SQLAlchemy column (e.g.: Post.metadata_).
        path: a list of keys (e.g.: ["seo", "title"]).
        value: any JSON-serializable value.
        create_missing: create the last key of the path if it doesn't exist.

    Returns:
        A SQL expression that produces the modified JSONB.
    """
    if not path:
        raise ValueError("path cannot be empty")
    pg_path = "{" + ",".join(path) + "}"
    json_value = json.dumps(value)
    return func.jsonb_set(column, pg_path, cast(json_value, JSONB), create_missing)


# Usage:
from sqlalchemy import update

stmt = (
    update(Post)
    .where(Post.id == 1)
    .values(metadata_=set_jsonb_path(Post.metadata_, ["seo", "title"], "New"))
)

UPDATE posts
SET metadata = jsonb_set(posts.metadata, '{seo,title}', '"New"', true)
WHERE posts.id = 1;

Exercise 4: a test that catches the bug

Write an async pytest test that catches the non-persisted mutations bug. The test must:

1. Create a post with metadata_={"views": 0}.
2. Modify it with post.metadata_["views"] = 99.
3. Commit.
4. Verify in a new session that views is 99.

# test_jsonb_persistence.py
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

# ... import Base and Post from your models module


@pytest.fixture
async def engine():
    engine = create_async_engine(
        "postgresql+asyncpg://postgres:postgres@localhost:5432/test",
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest.fixture
def session_factory(engine):
    return async_sessionmaker(engine, expire_on_commit=False)


@pytest.mark.asyncio
async def test_jsonb_mutation_persists(session_factory):
    # 1. Create the post
    async with session_factory() as s:
        post = Post(title="x", metadata_={"views": 0})
        s.add(post)
        await s.commit()
        post_id = post.id

    # 2 + 3. Modify and commit in one session
    async with session_factory() as s:
        post = (await s.execute(select(Post).where(Post.id == post_id))).scalar_one()
        post.metadata_["views"] = 99
        await s.commit()

    # 4. Verify in a new session
    async with session_factory() as s:
        post = (await s.execute(select(Post).where(Post.id == post_id))).scalar_one()
        assert post.metadata_["views"] == 99, (
            f"Change not persisted. metadata = {post.metadata_}. "
            "Did you forget MutableDict.as_mutable(JSONB)?"
        )

Exercise 5: an architectural decision

Your team has three endpoints that mutate posts.metadata:

a) POST /posts/{id}/views — increments views. Volume: 5000 req/s at peak hours. b) PATCH /posts/{id}/seo — updates seo with partial fields. Volume: 10 req/min, only from the internal CMS. c) A nightly job that recomputes metadata.last_indexed_at for all posts. Volume: 100k posts in one run, in batches.

For each one, say which pattern to use (MutableDict or func.jsonb_set) and why.

Summary and next step

In this capsule you learned the most expensive gotcha of JSONB with SQLAlchemy:

• The problem: SQLAlchemy detects reassignments (obj.x = y), not in-place mutations (obj.x["k"] = v). By default, mutating a dict loaded from JSONB does not trigger an UPDATE.
• MutableDict.as_mutable(JSONB): a wrapper that intercepts mutations of the root dict. Natural Python syntax, but only level 1 — for depth use reassignment or flag_modified. It rewrites the whole column on every commit.
• func.jsonb_set: an atomic update at the PostgreSQL level. It modifies only the indicated path. Better for high concurrency, large JSONs, and bulk updates. Verbose syntax, worth wrapping in a helper.
• Decision: MutableDict by default when it's comfortable and simple; func.jsonb_set when there's concurrency, a large JSON, or a batch.
• Critical tests: always verify persistence in a new session, not in the same one where you mutated. The bug is invisible if you only read the in-memory dict.

Before moving on you should be able to:

• Reproduce the bug and explain why SQLAlchemy doesn't detect it.
• Apply MutableDict.as_mutable(JSONB) to a column and know its limitation with nesting.
• Write an update with func.jsonb_set that modifies a specific path.
• Choose between the two patterns when facing a new endpoint.

Next capsule — Validation with Pydantic v2 and JSONB. So far your JSONB is dict[str, Any] — total flexibility, zero validation. As soon as an external client sends the endpoint a payload with og_image: 12345 (you expected a string, you got a number), your app crashes while rendering. Capsule 04 teaches you to validate incoming and outgoing JSONB with Pydantic v2 without sacrificing the flexibility JSONB gave you. You'll see TypeAdapter, nested schemas, and discriminated unions for polymorphic payloads — the foundation of capsule 07.

Resources

1. SQLAlchemy 2.0 — Mutation Tracking — the official reference for MutableDict, MutableList, and flag_modified.
2. SQLAlchemy 2.0 — func.jsonb_set and other JSON functions — the PostgreSQL dialect's function section.
3. PostgreSQL 16 — jsonb_set reference — the official documentation with all the options (create_missing, behavior with nonexistent paths).
4. Mike Bayer — "Why don't my changes to a JSON object persist?" — the ORM author's explanation of the change tracking model.
5. PostgreSQL 16 — Concurrency Control — the basis for why func.jsonb_set is atomic under MVCC.
6. pganalyze — JSONB updates and concurrency — an applied analysis of update patterns.

Module 2 — Advanced PostgreSQL for Backend Guide

Next capsule: Validation with Pydantic v2 and JSONB — turning a dict[str, Any] into validated data without losing flexibility.