Querying and indexing JSONB columns in SQLAlchemy

Map the column as Mapped[dict] = mapped_column(MutableDict.as_mutable(JSONB)), filter with Product.attributes.contains({"color": "red"}) so a GIN index can serve it, and use ["key"].astext or .as_integer() whenever you compare an extracted value to a Python scalar. This guide is part of querying PostgreSQL JSONB, arrays and full-text search.

Quick Answer

The most frequent JSONB mistakes are treating an extracted jsonb value as text, and editing a document in place without SQLAlchemy noticing. Both have one-line fixes.

Before — comparing jsonb to text, and an edit that is never saved:

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str]
    attributes: Mapped[dict] = mapped_column(JSONB, default=dict)


# attributes -> 'brand' is jsonb, not text, so text operators fail:
stmt = select(Product).where(Product.attributes["brand"].ilike("acme%"))
# ProgrammingError: operator does not exist: jsonb ~~* character varying

product.attributes["color"] = "blue"   # mutated in place
await session.commit()                 # no UPDATE is emitted

After — text extraction, containment, and tracked mutation:

from sqlalchemy import Index, select
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 Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str]
    attributes: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSONB), default=dict)

    __table_args__ = (
        Index("ix_products_attributes", "attributes", postgresql_using="gin"),
    )


# Text extraction for text operators and equality on one key:
by_brand = select(Product).where(Product.attributes["brand"].astext.ilike("acme%"))

# Containment, which the GIN index serves:
red = select(Product).where(Product.attributes.contains({"color": "red"}))

# Typed extraction for ranges:
in_stock = select(Product).where(Product.attributes["stock"].as_integer() > 0)

product.attributes["color"] = "blue"   # MutableDict marks the attribute dirty
await session.commit()                 # UPDATE products SET attributes=... WHERE id=...

Execution Context & Async Workflow Integration

SQLAlchemy's JSONB type does two separate jobs, and the bugs above come from confusing them. On the SQL side, its comparator turns Python indexing and method calls into PostgreSQL operators. On the Python side, its bind and result processors turn dictionaries into JSON text and back.

The operator mapping is precise. Product.attributes["brand"] renders attributes -> 'brand', which returns a jsonb value — the JSON string "Acme", quotes included. SQLAlchemy types that expression as JSON, so == "Acme" happens to work: the bound value is JSON-encoded too, and PostgreSQL compares two jsonb values. Everything text-shaped fails, though — ilike(), startswith(), a join to a VARCHAR column, a func.lower() — with operator does not exist: jsonb ~~* character varying or its relatives, and a jsonb-to-jsonb equality can never use an expression index on ->>. .astext renders ->> and returns text, which behaves like a string everywhere. For numbers and booleans, .as_integer(), .as_float() and .as_boolean() extract as text and cast, so > 0 is a numeric comparison rather than a lexical one.

Change tracking is a separate concern. The ORM detects changes by intercepting attribute assignment. product.attributes["color"] = "blue" does not assign to product.attributes; it mutates the dictionary the attribute already holds, so the unit of work sees nothing to flush. MutableDict.as_mutable(JSONB) wraps loaded values in a dictionary subclass that reports its own top-level changes to the parent. Nested changes — attributes["dims"]["width"] = 40 — are still invisible to it, and need either reassignment or flag_modified(product, "attributes").

Under asyncpg, serialisation happens in SQLAlchemy rather than in the driver: the asyncpg dialect registers a JSON codec that uses the engine's json_serializer and json_deserializer, which default to the standard library's json.dumps and json.loads. That is where TypeError: Object of type Decimal is not JSON serializable comes from, and it is raised at flush time, far from the assignment that caused it. The fix belongs on the engine, not on every assignment, and is shown in the advanced section below.

Because every JSONB write replaces the whole document, a large document edited frequently rewrites a large row frequently. That shapes both the upsert and conflict strategies you can use — a concurrent edit to a different key still conflicts — and the decision about which fields deserve real columns.

Resolving Warnings, Errors & Common Mistakes

The None row deserves a note, because it breaks IS NULL filters silently. A JSONB column with no none_as_null stores Python None as the JSON value null, which is not SQL NULL, so Product.attributes.is_(None) does not find it. Decide once which you mean: none_as_null=True if absence should be SQL NULL, or keep the default and filter with Product.attributes == JSON.NULL for explicitly-null documents.

Verify fixes with the query plan rather than timing. Reading EXPLAIN output for a SQLAlchemy query shows how to compile the exact statement and confirm that a Bitmap Index Scan on ix_products_attributes replaced the sequential scan.

Advanced: Indexing Strategy and Custom Serializers

A single default GIN index is a reasonable start, and it is not always the right end. Three index shapes cover almost every JSONB workload, and choosing between them is a matter of reading which operators your queries actually use.

from sqlalchemy import Index
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 Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str]
    attributes: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSONB), default=dict)

    __table_args__ = (
        # Containment only (@>), about a third smaller than the default opclass.
        Index(
            "ix_products_attributes_path",
            "attributes",
            postgresql_using="gin",
            postgresql_ops={"attributes": "jsonb_path_ops"},
        ),
    )


# Equality, ranges and ORDER BY on one extracted key. Declared after the class, because the
# expression needs the mapped attribute, not the mapped_column() placeholder.
Index("ix_products_brand", Product.attributes["brand"].astext)

The expression index only serves a query whose expression matches it exactly — attributes ->> 'brand', not attributes -> 'brand', and not lower(attributes ->> 'brand'). If a key is filtered, sorted and joined on constantly, that is a signal it has outgrown the document and wants to be a real column, possibly a generated one populated from the JSON.

Serialisation is configured once, on the engine. A serializer that handles the types your domain actually stores removes a whole class of flush-time errors:

import datetime as dt
import decimal
import json
import uuid

from sqlalchemy.ext.asyncio import create_async_engine


def _default(value):
    if isinstance(value, decimal.Decimal):
        return str(value)            # keep precision; parse back explicitly where needed
    if isinstance(value, (dt.datetime, dt.date)):
        return value.isoformat()
    if isinstance(value, uuid.UUID):
        return str(value)
    raise TypeError(f"cannot serialise {type(value).__name__}")


def dumps(value) -> str:
    return json.dumps(value, default=_default, separators=(",", ":"))


engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    json_serializer=dumps,
    json_deserializer=json.loads,
)

Serialising Decimal as a string rather than a float is deliberate: a float silently loses precision for money, and the string round-trips exactly. The cost is that the value comes back as a string, so code reading prices out of a document must convert it — another hint that prices belong in columns.

Advanced: jsonpath Filters and Unnesting Arrays in Documents

Documents often carry lists — tags, variant options, line items — and two questions come up about them: "does any element match?" and "give me one row per element". PostgreSQL answers the first with jsonpath and the second with a set-returning function, and SQLAlchemy 2.0 exposes both without dropping to text.

For matching inside nested structures, path_exists() renders the @? operator and path_match() renders @@. Both can use a GIN index built with either operator class:

from sqlalchemy import select

from shop.models import Product

# Any variant in stock with a price under 20:
cheap_in_stock = select(Product.sku).where(
    Product.attributes.path_exists("$.variants[*] ? (@.stock > 0 && @.price < 20)")
)

# Pass the threshold as a jsonpath variable instead of formatting it into the string:
from sqlalchemy import cast, func
from sqlalchemy.dialects.postgresql import JSONB

max_price = 20
parameterised = select(Product.sku).where(
    func.jsonb_path_exists(
        Product.attributes,
        "$.variants[*] ? (@.price < $max)",
        cast({"max": max_price}, JSONB),
    )
)

The second form matters for user input. A jsonpath expression is a small language, and building one with string formatting has the same problems as building SQL that way; jsonb_path_exists() accepts a vars document, so the value travels as a bound parameter.

For one row per element, a table-valued function in the FROM clause does the unnesting. column_valued() names the output so it can be selected, filtered and grouped like any column:

from sqlalchemy import func, select

from shop.models import Product

tag = func.jsonb_array_elements_text(Product.attributes["tags"]).column_valued("tag")

# Tag frequency across the catalogue:
tag_counts = (
    select(tag, func.count().label("products"))
    .group_by(tag)
    .order_by(func.count().desc())
    .limit(20)
)
# SELECT tag, count(*) AS products
# FROM products, jsonb_array_elements_text(products.attributes -> 'tags') AS tag
# GROUP BY tag ORDER BY count(*) DESC LIMIT 20

PostgreSQL treats a function in FROM that references an earlier table as an implicit lateral join, so each product contributes one row per tag. A product without a tags key contributes none, which is usually right for counting and wrong for a report that must list every product; for that, use an explicit LEFT JOIN LATERAL built with .lateral() and isouter=True.

If queries like these are the main way a list is read, it is worth asking whether the list wants to be a PostgreSQL ARRAY column or a child table instead — both are indexable in ways a list inside a document is not, and the parent guide compares the three.

Frequently Asked Questions

Related

• Querying Postgres JSONB, Arrays and Full-Text Search — The parent guide: JSONB, ARRAY and tsvector in one data model.
• Implementing full-text search with tsvector in SQLAlchemy — Generated tsvector columns, ranking and GIN indexes.
• Writing a TypeDecorator for encrypted columns — The TypeDecorator pattern for typed document access.
• Reading EXPLAIN output for a SQLAlchemy query — Confirming the GIN or expression index is used.