Cheatsheet for JSONB. Long-form: Postgres textbook Ch 2 .
Column
from sqlalchemy.dialects.postgresql import JSONB
class Event(Base):
__tablename__ = "events"
id: Mapped[int] = mapped_column(primary_key=True)
payload: Mapped[dict] = mapped_column(JSONB, nullable=False)
Always JSONB (binary, indexable), not JSON.
Queries
# Containment (uses GIN well)
stmt = select(Event).where(Event.payload.contains({"status": "ok"}))
# Path access
stmt = select(Event).where(Event.payload["user_id"].astext == "42")
# Cast to int
stmt = select(Event).where(Event.payload["user_id"].as_integer() == 42)
# Has key
from sqlalchemy import text
stmt = select(Event).where(Event.payload.has_key("error"))
# Multiple keys
stmt = select(Event).where(text("payload ?& array['a', 'b']")) # has all
stmt = select(Event).where(text("payload ?| array['a', 'b']")) # has any
# Deep path
stmt = select(Event).where(text("payload #>> '{user, email}' = :e"), {"e": "[email protected]"})
GIN index
__table_args__ = (
Index(
"ix_events_payload_gin",
"payload",
postgresql_using="gin",
postgresql_ops={"payload": "jsonb_path_ops"},
),
)
jsonb_path_ops: faster + smaller for @> containment. Use unless you need other operators.
Expression index for specific path
from sqlalchemy import Index, text
__table_args__ = (
Index("ix_events_user_id", text("(payload->>'user_id')")),
)
Smaller than full GIN; targeted.
Updates
from sqlalchemy import func, update
# Concat (shallow merge)
await session.execute(
update(User).where(User.id == 1).values(prefs=User.prefs + {"theme": "dark"})
)
# Set a nested path
await session.execute(
update(User)
.where(User.id == 1)
.values(prefs=func.jsonb_set(User.prefs, "{notifications, email}", "true"))
)
# Remove a key (use #- for path)
await session.execute(
update(User).where(User.id == 1).values(prefs=text("prefs - 'theme'"))
)
Filtering on JSONB array elements
# Array contains object
stmt = select(Order).where(text("orders.items @> :v::jsonb"), {"v": '[{"sku": "ABC"}]'})
# jsonb_path_query (powerful path API)
from sqlalchemy import literal_column
stmt = select(Order).where(
text("jsonb_path_exists(items, '$ ? (@.qty > 5)')")
)
ON CONFLICT DO UPDATE with JSONB
from sqlalchemy.dialects.postgresql import insert as pg_insert
stmt = pg_insert(User).values(id=1, prefs={"theme": "dark"})
stmt = stmt.on_conflict_do_update(
index_elements=["id"],
set_={"prefs": User.prefs + stmt.excluded.prefs}, # merge
)
await session.execute(stmt)
Custom Pydantic-backed JSONB
from sqlalchemy.types import TypeDecorator
from pydantic import BaseModel
class PydanticJSONB(TypeDecorator):
impl = JSONB
cache_ok = True
def __init__(self, model_cls, **kw):
self.model_cls = model_cls; super().__init__(**kw)
def process_bind_param(self, value, dialect):
if value is None: return None
return value.model_dump() if isinstance(value, BaseModel) else value
def process_result_value(self, value, dialect):
if value is None: return None
return self.model_cls.model_validate(value)
class Config(BaseModel):
theme: str = "light"
class User(Base):
config: Mapped[Config] = mapped_column(PydanticJSONB(Config), default=Config)
Schema validation (pg_jsonschema)
CREATE EXTENSION pg_jsonschema;
__table_args__ = (
CheckConstraint(
"jsonschema_is_valid('{\"type\":\"object\",\"required\":[\"user_id\"]}'::json, payload)",
name="ck_events_payload_valid",
),
)
Common JSONB anti-patterns
- Storing first-class relational data as JSONB (lose type safety, joins).
- Deep nesting beyond 2-3 levels (hard to query, indexed).
- Huge JSONB blobs (>1MB) — TOAST overhead, slow updates.
- Forgetting GIN — seq scans on millions of rows.
Common mistakes
JSONinstead ofJSONB.- GIN without
jsonb_path_opsopclass. - Index expression doesn’t match query expression (cast mismatch).
- Updating a single nested key by replacing entire JSONB blob.
Read this next
If you want my Pydantic-JSONB + schema-validation patterns, it’s at rajpoot.dev .
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .