`Mapped[dict]` and JSONB types in SQLAlchemy 2.0

Capsule overview

In module 1 you mastered JSONB in pure SQL. Now you use it from Python. What seems trivial — declaring the column, reading a field, filtering by containment — has nuances that separate code that works in a demo from code that survives a serious code review: which default you use (default=dict, not default={}), which type hint you declare (dict[str, Any], not dict), which accessor you write (Post.metadata["seo"]["title"].astext vs func.jsonb_extract_path_text(...)), and what SQL the ORM ends up executing.

This capsule gives you the complete catalog. You learn to declare JSONB columns in SQLAlchemy 2.0 models with correct type hints, to access nested fields with ORM syntax, to filter with JSONB.contains() and JSONB.has_key(), and to use func.jsonb_extract_path_text when the ORM accessors fall short. Every example shows the SQL SQLAlchemy generates, because without that visibility the ORM becomes magic and debugging becomes guesswork.

By the end you'll be able to write the SQLAlchemy equivalent of any JSONB query you learned in module 1, without googling, and predict the SQL that will come out.

Mental model: the ORM as a translator, not an abstraction

The most important thing you can internalize about SQLAlchemy + JSONB is this: the ORM doesn't hide the SQL, it translates it. Every accessor, every .contains(), every func.X corresponds one-to-one with a fragment of SQL you'll be able to predict.

Python (SQLAlchemy)                         Generated SQL
──────────────────                          ────────────
Post.metadata["seo"]                   →    posts.metadata -> 'seo'
Post.metadata["seo"]["title"]          →    posts.metadata -> 'seo' -> 'title'
Post.metadata["seo"]["title"].astext   →    posts.metadata -> 'seo' ->> 'title'
Post.metadata.contains({"x": 1})       →    posts.metadata @> '{"x": 1}'
Post.metadata.has_key("x")             →    posts.metadata ? 'x'
func.jsonb_extract_path_text(
  Post.metadata, "seo", "title"
)                                      →    jsonb_extract_path_text(
                                              posts.metadata, 'seo', 'title'
                                            )

Memorize that table. Every time you're unsure what SQL an accessor produces, come back to it. And when you write new code, keep the habit of running with echo=True or reading the query log until the SQL feels obvious.

Why it matters: a badly written query in SQLAlchemy can generate SQL that doesn't use the GIN you created. The planner sees posts.metadata ->> 'status' = 'active' and does a sequential scan; you expected @> and your code review approved something that doesn't scale. If you know the SQL your code produces, you avoid those mistakes before the first commit.

Declaring a JSONB column in SQLAlchemy 2.0

The modern SQLAlchemy 2.0 style uses Mapped[T] and mapped_column. For JSONB in PostgreSQL you import the type from the dialect:

# models.py
from typing import Any
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy import BigInteger


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    body: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",          # column name in SQL (renamed so it doesn't clash with DeclarativeBase.metadata)
        JSONB,
        nullable=False,
        default=dict,
    )

Three details that seem minor and aren't:

1. default=dict, not default={}

# WRONG
default={}

# RIGHT
default=dict

default={} evaluates the empty dictionary only once, when the class is defined. Every instance would share the same object. If one modifies the dict in-place, the others would see the change. It's Python's classic "mutable default arguments" bug.

default=dict (without parentheses) passes the function to SQLAlchemy. The ORM invokes it for each new instance, generating a fresh dict. That's the correct way.

2. Mapped[dict[str, Any]], not Mapped[dict]

dict without parameters tells Python "any dict." dict[str, Any] documents that the keys are strings (which JSONB requires) and the values are anything serializable to JSON. It helps the type checker (mypy/pyright) and whoever reads the code.

If your JSONB has a known shape, you can be stricter with TypedDict or a Pydantic model (we'll see it in capsule 04).

3. metadata_ with a rename

DeclarativeBase already has a class attribute called metadata (SQLAlchemy's MetaData). If you name your column metadata, you're going to collide. The convention is to call the Python attribute metadata_ (with an underscore) and rename it to the real SQL name with the first positional argument of mapped_column.

If your entity doesn't use the name metadata, you don't have this problem:

class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)

payload doesn't clash with anything — straightforward.

Minimal Alembic migration

To create this table with Alembic, after alembic revision --autogenerate -m "create posts", the generated script looks like this:

# alembic/versions/xxxx_create_posts.py
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql


def upgrade() -> None:
    op.create_table(
        "posts",
        sa.Column("id", sa.BigInteger(), primary_key=True),
        sa.Column("title", sa.String(), nullable=False),
        sa.Column("body", sa.String(), nullable=False),
        sa.Column(
            "metadata",
            postgresql.JSONB(astext_type=sa.Text()),
            nullable=False,
            server_default=sa.text("'{}'::jsonb"),
        ),
    )


def downgrade() -> None:
    op.drop_table("posts")

Note the server_default=sa.text("'{}'::jsonb"). default=dict in the model applies from Python; server_default applies from PostgreSQL's side for rows inserted by pure SQL or by another app that doesn't go through the ORM. If your table has pre-existing data and you add the column afterwards, the server_default avoids an initial NULL.

Inserting and reading JSONB

Once the column is declared, writing and reading is straightforward. An end-to-end example with async SQLAlchemy 2.0:

# example_insert_read.py
import asyncio
from typing import Any

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


class Base(DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    title: Mapped[str]
    body: 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,  # prints the SQL — leave it on while learning
    )
    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)

    # INSERT
    async with Session() as session:
        post = Post(
            title="JSONB with SQLAlchemy",
            body="Lorem ipsum",
            metadata_={
                "seo": {"title": "JSONB for Python devs", "description": "Guide"},
                "tags": ["postgres", "sqlalchemy"],
                "published": True,
            },
        )
        session.add(post)
        await session.commit()
        post_id = post.id

    # SELECT
    async with Session() as session:
        result = await session.execute(select(Post).where(Post.id == post_id))
        loaded = result.scalar_one()
        print(loaded.metadata_)
        print(loaded.metadata_["seo"]["title"])

    await engine.dispose()


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

Run it:

python example_insert_read.py

Expected output (abridged):

-- What SQLAlchemy executes (thanks to echo=True):
INSERT INTO posts (title, body, metadata) VALUES ($1::VARCHAR, $2::VARCHAR, $3)
-- with parameters: ('JSONB with SQLAlchemy', 'Lorem ipsum', '{"seo": {"title": "JSONB for Python devs", ...}}')

SELECT posts.id, posts.title, posts.body, posts.metadata
FROM posts
WHERE posts.id = $1::INTEGER

# What Python prints:
{'seo': {'title': 'JSONB for Python devs', 'description': 'Guide'}, 'tags': ['postgres', 'sqlalchemy'], 'published': True}
JSONB for Python devs

Note: SQLAlchemy serializes the Python dict to JSON automatically on insert and deserializes it on read. You don't have to call json.dumps/json.loads manually.

Accessing nested fields with ORM accessors

You already know that in pure SQL the access is data->'k', data->>'k', data#>'{a,b}', data#>>'{a,b}'. SQLAlchemy exposes these operators with Python syntax:

Access to a key

# Python
Post.metadata_["seo"]            # → metadata -> 'seo'    (returns JSONB)
Post.metadata_["seo"].astext     # → metadata ->> 'seo'   (returns text)

obj["k"] (subscript) is the equivalent of ->. If you want the result as text (the most common case for comparing against strings or returning to the app), chain .astext.

Nested access

# Chained (equivalent to -> -> -> -> -> ->>)
Post.metadata_["seo"]["title"].astext   # → metadata -> 'seo' -> 'title' ->> ... 
                                        # (actually: metadata #>> '{seo,title}')

SQLAlchemy is smart: when you chain ["k1"]["k2"]...["kn"].astext, internally it translates to #>> with a path. You'll see metadata #>> '{seo,title}' in the SQL. Cleaner than ->.->.->>.

Filtering in WHERE

# SELECT * FROM posts WHERE metadata #>> '{seo, title}' = 'JSONB for Python devs'
stmt = select(Post).where(
    Post.metadata_["seo"]["title"].astext == "JSONB for Python devs"
)

Comparing numbers — cast explicitly

->> and .astext return text. If you compare against numbers without casting, you compare strings (lexicographically):

from sqlalchemy import cast, Integer

# WRONG: compares strings
select(Post).where(Post.metadata_["views"].astext > "100")
# "99" > "100" is True by lexicographic order — a latent bug.

# RIGHT
select(Post).where(cast(Post.metadata_["views"].astext, Integer) > 100)

SQLAlchemy's cast generates (metadata #>> '{views}')::INTEGER in SQL.

Search operators as methods

The search operators from module 1 (@>, ?, ?|, ?&) are exposed as methods of the JSONB type:

JSONB.contains() — the equivalent of @>

# SELECT * FROM posts WHERE metadata @> '{"published": true}'
stmt = select(Post).where(Post.metadata_.contains({"published": True}))

# With a nested sub-object
stmt = select(Post).where(Post.metadata_.contains({"seo": {"title": "X"}}))

contains is the most important method in the whole capsule — it's the one the GIN from module 1 speeds up. Whenever you can rewrite a filter as Post.metadata_.contains({...}), do it. It's the SQLAlchemy version of the @> you already know.

JSONB.contained_by() — the equivalent of <@

# SELECT * FROM tags_table WHERE tags <@ '["python", "postgres", "sqlalchemy"]'
stmt = select(TagsRow).where(TagsRow.tags.contained_by(["python", "postgres", "sqlalchemy"]))

Less common. Useful when you ask "is this JSONB a subset of that other one?"

JSONB.has_key() — the equivalent of ?

# SELECT * FROM posts WHERE metadata ? 'seo'
stmt = select(Post).where(Post.metadata_.has_key("seo"))

JSONB.has_any() and JSONB.has_all() — the equivalents of ?| and ?&

# SELECT * FROM users WHERE settings ?| ARRAY['feature_a', 'feature_b']
from sqlalchemy import cast, ARRAY, String
# The has_any method expects the right-hand side as an array of strings.
stmt = select(User).where(
    User.settings.has_any(cast(["feature_a", "feature_b"], ARRAY(String)))
)

# A simpler version with func:
from sqlalchemy import func
stmt = select(User).where(
    func.jsonb_exists_any(User.settings, ["feature_a", "feature_b"])
)

If the JSONB wrappers feel awkward to you for edge cases, you can always fall back to func.<postgres_function> directly. Both options generate the same SQL.

When the accessors fall short: func.jsonb_extract_path_text

The ["k"]["k"].astext accessors cover 90% of cases. The remaining 10%:

• A path with keys that come from Python variables.
• Very deep JSONs where chaining becomes unreadable.
• When you want to be explicit about the SQL function you're calling.

For these, use func.jsonb_extract_path_text (returns text) or func.jsonb_extract_path (returns jsonb):

from sqlalchemy import func

# Equivalent to metadata #>> '{seo,title}'
stmt = select(Post).where(
    func.jsonb_extract_path_text(Post.metadata_, "seo", "title") == "X"
)

# Path from variables
path_keys = ["seo", "open_graph", "image"]
stmt = select(Post).where(
    func.jsonb_extract_path_text(Post.metadata_, *path_keys) == "https://..."
)

It generates SQL identical to #>>:

SELECT posts.id, posts.title, ...
FROM posts
WHERE jsonb_extract_path_text(posts.metadata, 'seo', 'open_graph', 'image') = 'https://...'

Worked example: an end-to-end query with accessors and contains

The case: in your Blog API, you want to list every published post (metadata.published = true), from the engineering category (metadata.category = "engineering"), sorted by views descending, returning the title, slug, and canonical URL from the metadata.

# example_query.py
import asyncio
from typing import Any

from sqlalchemy import BigInteger, cast, Integer, select, desc
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]
    slug: Mapped[str]
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata", JSONB, nullable=False, default=dict
    )


async def seed(session) -> None:
    posts = [
        Post(
            title="Async with FastAPI",
            slug="async-fastapi",
            metadata_={
                "published": True,
                "category": "engineering",
                "views": 5400,
                "seo": {"canonical": "https://blog.example.com/async-fastapi"},
            },
        ),
        Post(
            title="How to organize a remote team",
            slug="remote-team",
            metadata_={
                "published": True,
                "category": "management",
                "views": 1200,
                "seo": {"canonical": "https://blog.example.com/remote-team"},
            },
        ),
        Post(
            title="Draft",
            slug="draft-1",
            metadata_={"published": False, "category": "engineering", "views": 0},
        ),
        Post(
            title="JSONB in production",
            slug="jsonb-prod",
            metadata_={
                "published": True,
                "category": "engineering",
                "views": 8900,
                "seo": {"canonical": "https://blog.example.com/jsonb-prod"},
            },
        ),
    ]
    session.add_all(posts)
    await session.commit()


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)
    async with Session() as session:
        await seed(session)

    async with Session() as session:
        stmt = (
            select(
                Post.title,
                Post.slug,
                Post.metadata_["seo"]["canonical"].astext.label("canonical"),
            )
            .where(
                Post.metadata_.contains(
                    {"published": True, "category": "engineering"}
                )
            )
            .order_by(
                desc(cast(Post.metadata_["views"].astext, Integer))
            )
        )
        result = await session.execute(stmt)
        for row in result.all():
            print(row.title, "|", row.slug, "|", row.canonical)

    await engine.dispose()


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

The SQL SQLAlchemy generates (with echo=True):

SELECT
  posts.title,
  posts.slug,
  (posts.metadata #>> '{seo,canonical}') AS canonical
FROM posts
WHERE posts.metadata @> '{"category": "engineering", "published": true}'
ORDER BY (posts.metadata #>> '{views}')::INTEGER DESC

Expected output:

JSONB in production | jsonb-prod | https://blog.example.com/jsonb-prod
Async with FastAPI | async-fastapi | https://blog.example.com/async-fastapi

What's happening:

1. Post.metadata_.contains({...}) generates metadata @> '{...}'. That's the part that's indexable with GIN.
2. Post.metadata_["seo"]["canonical"].astext generates metadata #>> '{seo,canonical}'. We project it in the SELECT with .label("canonical").
3. cast(Post.metadata_["views"].astext, Integer) generates (metadata #>> '{views}')::INTEGER. Without the cast, you'd sort as a string (5400 < 8900 happens to work, but 9 > 1000 doesn't). Always cast.

Why does this matter in real work?

1. The SQL that comes out of the ORM determines whether your GIN gets used. If you declare a GIN over posts.metadata and filter with Post.metadata_["status"].astext == "published", the GIN isn't used (it's an access operator, not a search one). If you filter with Post.metadata_.contains({"status": "published"}), it is. Same intent, different performance. Knowing how to translate Python to SQL saves you from that mistake.

2. Stable accessors let you refactor without breaking queries. If you decide to change metadata.seo.title to metadata.seo.h1 for all posts, you modify the accessor in a single place. If your code had 30 hardcoded strings with metadata->>'seo'->>'title', refactoring is a risky global find/replace. Type hints + accessors give you IDE-assisted refactoring.

3. func.jsonb_extract_path_text is your tool for dynamic queries. When a filter comes from the user (?filter[metadata.tags]=python), you can't hardcode the accessor — you have to build it from a string. func.jsonb_extract_path_text(col, *path_parts) is what you use.

Traps and common mistakes

Mistake 1 (conceptual): assuming ["k"] returns a Python dict

Symptom: you write Post.metadata_["seo"] expecting it to return the sub-dict and you do Post.metadata_["seo"]["title"] in ["X", "Y"]. It fails.

Why it happens: Post.metadata_["seo"] doesn't return a dict — it returns a SQL expression (a SQLAlchemy BinaryExpression object). You're building SQL, not manipulating data.

How to tell them apart: if you're inside a select(...).where(...), everything you write is SQL expressions. If you're working with a loaded object (post = result.scalar_one()), then post.metadata_["seo"] really is a Python dict.

Fix: to use IN in SQL, use .in_():

stmt = select(Post).where(Post.metadata_["seo"]["title"].astext.in_(["X", "Y"]))

Mistake 2 (conceptual): forgetting .astext and comparing jsonb with a string

Symptom: Post.metadata_["seo"]["title"] == "Hello" doesn't return rows even though the title is exactly "Hello".

Why it happens: without .astext, the accessor returns jsonb, not text. SQLAlchemy compares metadata -> 'seo' -> 'title' = 'Hello' (jsonb vs text), which in PostgreSQL isn't a valid direct comparison.

Fix: always .astext before comparing against Python strings:

Post.metadata_["seo"]["title"].astext == "Hello"

Mistake 3 (practical): using default={} instead of default=dict

Symptom: two posts share metadata without you having touched them. Modifying one's shows up in the other.

Why it happens: default={} evaluates the dict only once when the class is loaded. Every new instance receives the same object as its default.

Fix: default=dict. SQLAlchemy invokes the function for each new instance and each one gets its own dict.

Mistake 4 (practical): numeric comparison without a cast

Symptom: select(Post).where(Post.metadata_["views"].astext > "1000") returns rows with views=99 (because "99" > "1000" as strings).

Fix: always cast to the numeric type:

from sqlalchemy import cast, Integer
select(Post).where(cast(Post.metadata_["views"].astext, Integer) > 1000)

And if that query runs often, consider an expression index:

CREATE INDEX posts_views_idx ON posts (((metadata ->> 'views')::int));

Mistake 5 (conceptual): expecting the metadata collision not to matter if you use __init__

Symptom: you declare metadata: Mapped[dict] = ... in a class that inherits from DeclarativeBase and you get strange errors about MetaData.

Why it happens: DeclarativeBase uses the metadata attribute for its own internal MetaData. Your column overwrites it.

Fix: rename the Python attribute (metadata_, meta, extra, etc.) and use the first argument of mapped_column to keep metadata as the SQL name if you need it:

metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSONB, default=dict)

Exercises

Exercise 1: translate SQL to SQLAlchemy

You have the following SQL:

SELECT id, payload ->> 'event_type' AS event_type
FROM events
WHERE payload @> '{"action": "purchase", "country": "US"}'
  AND (payload ->> 'amount')::numeric > 50
ORDER BY (payload ->> 'amount')::numeric DESC;

Assume the model:

class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)

Write the equivalent query in SQLAlchemy 2.0.

from sqlalchemy import select, cast, Numeric, desc

stmt = (
    select(
        Event.id,
        Event.payload["event_type"].astext.label("event_type"),
    )
    .where(
        Event.payload.contains({"action": "purchase", "country": "US"}),
        cast(Event.payload["amount"].astext, Numeric) > 50,
    )
    .order_by(desc(cast(Event.payload["amount"].astext, Numeric)))
)

Exercise 2: predict the SQL

Without running it, what SQL does the following code generate? Is it indexable with a GIN over payload?

stmt = select(Event).where(
    Event.payload["device"].astext == "mobile",
    Event.payload.has_key("campaign_id"),
)

SELECT events.id, events.payload
FROM events
WHERE (events.payload ->> 'device') = 'mobile'
  AND (events.payload ? 'campaign_id');

stmt = select(Event).where(
    Event.payload.contains({"device": "mobile"}),
    Event.payload.has_key("campaign_id"),
)

Exercise 3: deep extraction and a combined filter

The model:

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    profile: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)

Example data:

users = [
    User(profile={
        "name": "Ana",
        "address": {"country": "MX", "city": "CDMX"},
        "preferences": {"theme": "dark", "newsletter": True},
    }),
    User(profile={
        "name": "Luis",
        "address": {"country": "AR", "city": "Buenos Aires"},
        "preferences": {"theme": "light", "newsletter": False},
    }),
]

Write a query that returns the name and the city of the users from Mexico who have the newsletter enabled.

from sqlalchemy import select

stmt = select(
    User.profile["name"].astext.label("name"),
    User.profile["address"]["city"].astext.label("city"),
).where(
    User.profile.contains({
        "address": {"country": "MX"},
        "preferences": {"newsletter": True},
    })
)

SELECT
  (users.profile ->> 'name') AS name,
  (users.profile #>> '{address,city}') AS city
FROM users
WHERE users.profile @> '{"address": {"country": "MX"}, "preferences": {"newsletter": true}}';

Ana | CDMX

Exercise 4: fix a numeric comparison bug

The following code has a bug. Identify it and fix it.

stmt = select(Event).where(
    Event.payload["score"].astext > "100"
)

from sqlalchemy import cast, Numeric

stmt = select(Event).where(
    cast(Event.payload["score"].astext, Numeric) > 100
)

SELECT * FROM events WHERE (events.payload ->> 'score')::NUMERIC > 100;

CREATE INDEX events_score_idx ON events (((payload ->> 'score')::numeric));

Exercise 5: a dynamic query with a path from variables

You're building a GET /posts/search endpoint that receives a field parameter indicating a path in metadata and a value to compare. For example, field=seo.title&value=Hello searches for posts where metadata #>> '{seo,title}' = 'Hello'.

Write the Python function that takes field: str and value: str and returns the corresponding SQLAlchemy query, validating that the field isn't empty.

from sqlalchemy import select, func

def build_search_query(field: str, value: str):
    if not field:
        raise ValueError("field is required")
    path_parts = field.split(".")
    if not all(part for part in path_parts):
        raise ValueError("field has empty segments")
    return select(Post).where(
        func.jsonb_extract_path_text(Post.metadata_, *path_parts) == value
    )

SELECT posts.id, posts.title, posts.body, posts.metadata
FROM posts
WHERE jsonb_extract_path_text(posts.metadata, 'seo', 'title') = 'Hello';

Exercise 6: distinguish contains from has_key

For each case, say which SQLAlchemy method to use and what SQL it produces:

a) "Posts that have any metadata under the seo key." b) "Posts whose seo.title is defined (not null and not absent)." c) "Posts where category is exactly engineering or product." d) "Posts where metadata contains either of the top-level keys featured or pinned."

from sqlalchemy import or_
or_(
    Post.metadata_.contains({"category": "engineering"}),
    Post.metadata_.contains({"category": "product"}),
)

Summary and next step

In this capsule you learned to express JSONB from SQLAlchemy 2.0:

• Declaration: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict). default=dict, not default={}. If your attribute is called metadata, rename it (it collides with DeclarativeBase).
• Accessors: obj["k"] is ->, obj["k"].astext is ->>, obj["a"]["b"].astext is #>> with a path.
• Search methods: contains({...}) is @> (GIN-indexable), has_key("k") is ?, has_any/has_all are ?|/?&.
• Always cast when comparing numbers: cast(col["x"].astext, Integer) > 100.
• For dynamic paths from variables, func.jsonb_extract_path_text(col, *parts).
• The SQL that comes out of the ORM determines whether your GIN gets used. contains leverages it; ["k"].astext == "x" doesn't.

Before moving on you should be able to:

• Declare a JSONB column without googling.
• Predict the SQL a chained accessor and contains generate.
• Rewrite a filter with ["k"].astext == "x" to contains({"k": "x"}) when it suits the GIN.
• Cast correctly for numeric comparisons.

Next capsule — Mutations and MutableDict. So far you've read and filtered. When you try to modify the dict — post.metadata_["views"] = 1234 and then await session.commit() — you'll discover that SQLAlchemy doesn't detect the change and doesn't generate an UPDATE. That's the most expensive gotcha of JSONB with an ORM. Capsule 03 teaches you the two patterns for fixing it: MutableDict.as_mutable(JSONB) for in-place mutations and func.jsonb_set for partial updates without rewriting the whole column.

Resources

1. SQLAlchemy 2.0 — JSONB reference — all the methods of the type (contains, has_key, has_any, has_all).
2. SQLAlchemy 2.0 — JSON Operators (Index/Path Access) — explains how ["k"] and .astext translate to SQL.
3. SQLAlchemy 2.0 — Mapped and mapped_column — the modern style for declaring columns.
4. PostgreSQL 16 — JSON Functions and Operators — the SQL reference for jsonb_extract_path, jsonb_extract_path_text, etc.
5. Mike Bayer — Asynchronous I/O with SQLAlchemy — the official guide to the async style used in the examples.
6. Alembic — Autogenerate Migrations — how to generate migrations from models with JSONB.

Module 2 — Advanced PostgreSQL for Backend Guide

Next capsule: Mutations and MutableDict — the gotcha that loses your changes silently.