Module 2: JSONB with SQLAlchemy and usage patterns

Pattern: extensible metadata with JSONB

Capsule description

The second canonical JSONB pattern. While "dynamic configuration" (capsule 05) is per-entity and read by the app (tenants.settings), extensible metadata is per-instance and stores information that's part of the resource but whose schema evolves with your customers and use cases: a post's SEO, custom fields that each SaaS customer defines to their own taste, per-category product attributes, per-event integration data.

This capsule teaches you how to model metadata that grows without migrations, how to combine known fields (validated with Pydantic) with custom fields (extensible), how to build the smell test that separates "healthy metadata" from "metadata that's become a badly designed mini-database," and how to apply all of it to the posts.metadata that will show up in this module's final project (capsule 08) and in the guide's integrative project.

By the end you'll be able to tell when a new field on your model goes in metadata vs in a new column, and you'll be able to defend the decision with concrete criteria when a teammate asks "why isn't it a column?"


Mental model: the difference between config and metadata

Even though both live in a JSONB column, they have different purposes:

┌───────────────────────────────────┬─────────────────────────────────────┐
│  CONFIGURATION (capsule 05)       │  METADATA (this capsule)            │
│                                   │                                     │
│  Who defines it: your team        │  Who defines it: your end customer  │
│  (product, engineering)           │  (who knows the domain)             │
│                                   │                                     │
│  When it changes: every sprint,   │  When it changes: every use, for    │
│  by your action                   │  each individual resource           │
│                                   │                                     │
│  Schema: fairly well known        │  Schema: partially known            │
│  (you define the flags upfront)   │  (some fixed fields, others         │
│                                   │  depend on the customer)            │
│                                   │                                     │
│  Reads: the code makes            │  Reads: the code displays info,     │
│  decisions (if has_feature)       │  the customer consumes it           │
│                                   │                                     │
│  Typical size: <5 KB              │  Typical size: <10 KB               │
│                                   │                                     │
│  Examples: feature flags, theme,  │  Examples: SEO, custom fields,      │
│  integrations                     │  product attributes, webhook        │
│                                   │  payload                            │
└───────────────────────────────────┴─────────────────────────────────────┘

The distinction matters because:

  • Validation. Config has strict Pydantic (your team controls the shape). Metadata has Pydantic with extra="allow" (customers add fields you didn't anticipate).
  • Migration. You version config with your app (a release adds feature_y). Metadata evolves naturally without coordination.
  • Queries. Config gets filtered with @> for "tenants with flag X." Metadata gets projected more than filtered ("give me this post's custom fields").

The anchor case: posts.metadata for SEO and custom fields

You're going to model the metadata column of the posts table that the final project will use. Three categories of information live together:

  1. Standard SEO: title, description, canonical, og_image. Known fields your team added to the product.
  2. Per-customer custom fields: one customer wants legal_disclaimer, another wants affiliate_id, another wants experiment_variant. Your team doesn't know them all.
  3. Computed/auxiliary data: last_indexed_at, read_time_minutes, word_count. Generated by internal jobs.

Schema:

# models.py
from datetime import datetime
from typing import Any, Literal

from sqlalchemy import BigInteger, Index, String, Text
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] = mapped_column(String(200))
    slug: Mapped[str] = mapped_column(String(200), unique=True)
    body: Mapped[str] = mapped_column(Text)
    status: Mapped[Literal["draft", "published", "archived"]] = mapped_column(
        default="draft"
    )
    published_at: Mapped[datetime | None]

    # Extensible metadata: SEO + custom + computed
    metadata_: Mapped[dict[str, Any]] = mapped_column(
        "metadata",
        MutableDict.as_mutable(JSONB),
        nullable=False,
        default=dict,
    )

    __table_args__ = (
        Index("ix_posts_metadata_gin", "metadata", postgresql_using="gin"),
    )

Schema decisions:

  • title, slug, body, status, published_at are typed columns. They're the Post's core model: listing queries, search, ordering. Stable, with constraints (slug unique, status enum, published_at nullable).
  • metadata_ is JSONB. Everything that evolves with each customer and case goes here.
  • The GIN on metadata allows fast @> queries (e.g. "every post with a specific seo.canonical"). For this case, jsonb_path_ops would be enough if you only use @>. If you're going to use ? ("which posts have seo defined?"), the default jsonb_ops applies. That decision is covered in module 1, capsule 05.

The Pydantic schema with three levels of extensibility

# pydantic_metadata.py
from datetime import datetime
from pydantic import BaseModel, Field, HttpUrl


class PostSEO(BaseModel):
    """Standard SEO: known fields, strict validation."""
    title: str = Field(min_length=1, max_length=160)
    description: str = Field(min_length=1, max_length=320)
    canonical: HttpUrl
    og_image: HttpUrl | None = None
    twitter_card: str | None = Field(default=None, max_length=50)


class PostComputed(BaseModel):
    """Data computed by internal jobs. Schema controlled by your team."""
    word_count: int = Field(ge=0, default=0)
    read_time_minutes: int = Field(ge=0, default=0)
    last_indexed_at: datetime | None = None


class PostMetadata(BaseModel):
    """
    General metadata structure.
    - seo: optional but strictly validated.
    - computed: optional but strictly validated.
    - custom_fields: a free dict with the customer's keys/values.
    - top-level extras: allowed for future expansion.
    """
    seo: PostSEO | None = None
    computed: PostComputed | None = None
    custom_fields: dict[str, str | int | bool | None] = Field(default_factory=dict)

    model_config = {"extra": "allow"}

Three levels of "strictness":

  1. seo and computed: known schemas, full validation. If the marketing team asks for seo.title, a customer who sends seo: {"title": null} gets a 422. It guarantees the contract.

  2. custom_fields: a dict with arbitrary keys ({"legal_disclaimer": "...", "affiliate_id": "abc"}) but typed values (string, int, bool, null). It allows controlled extensibility — the customer adds fields but can't stuff in an arbitrarily deep nested object that breaks the app.

  3. extra="allow" at the top level: if your team decides in the future to add metadata.experiment_results, it doesn't break what exists. New keys that don't have a Pydantic schema yet simply pass through.

It's flexibility in layers: what matters is strictly validated, the customer's stuff has a basic shape, the future is allowed.


Typical operations

1. Create a post with mixed metadata

from fastapi import FastAPI
from pydantic import BaseModel


class CreatePostRequest(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    slug: str = Field(min_length=1, max_length=200, pattern=r"^[a-z0-9-]+$")
    body: str = Field(min_length=1)
    metadata: PostMetadata = Field(default_factory=PostMetadata)


app = FastAPI()


@app.post("/posts", status_code=201)
async def create_post(payload: CreatePostRequest) -> dict[str, Any]:
    async with SessionLocal() as session:
        post = Post(
            title=payload.title,
            slug=payload.slug,
            body=payload.body,
            metadata_=payload.metadata.model_dump(mode="json"),
        )
        session.add(post)
        await session.commit()
        return {"id": post.id, "slug": post.slug}

The client sends:

{
  "title": "How to use JSONB",
  "slug": "how-to-use-jsonb",
  "body": "Lorem ipsum",
  "metadata": {
    "seo": {
      "title": "JSONB guide for Python devs",
      "description": "Learn JSONB with SQLAlchemy",
      "canonical": "https://blog.example.com/how-to-use-jsonb"
    },
    "custom_fields": {
      "affiliate_id": "abc123",
      "experiment_variant": "v2"
    }
  }
}

Pydantic validates seo strictly (it rejects the request if description is missing), allows custom_fields with any key, and stores it all in JSONB.

2. Update one SEO field without touching the rest

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


async def update_seo_title(session, post_id: int, new_title: str) -> None:
    """
    Partial update: only metadata.seo.title changes.
    The rest of metadata (custom_fields, computed, other fields)
    is preserved untouched.
    """
    stmt = (
        update(Post)
        .where(Post.id == post_id)
        .values(
            metadata_=func.jsonb_set(
                Post.metadata_,
                "{seo,title}",
                cast(json.dumps(new_title), JSONB),
                True,  # creates seo.title if seo exists but title doesn't
            )
        )
    )
    await session.execute(stmt)
    await session.commit()

Why func.jsonb_set and not MutableDict:

  • If two editors change different fields of the same post simultaneously, MutableDict rewrites the entire column and one clobbers the other. func.jsonb_set only modifies the given path.
  • For a blog's SEO field, with low concurrency, MutableDict could work. For real production, jsonb_set is better.

3. Add a custom field

async def add_custom_field(
    session, post_id: int, key: str, value: str | int | bool | None
) -> None:
    if not key or "," in key or "{" in key:
        raise ValueError("invalid key for jsonb path")
    stmt = (
        update(Post)
        .where(Post.id == post_id)
        .values(
            metadata_=func.jsonb_set(
                func.jsonb_set(
                    Post.metadata_,
                    "{custom_fields}",
                    cast("{}", JSONB),
                    True,
                ),
                "{custom_fields," + key + "}",
                cast(json.dumps(value), JSONB),
                True,
            )
        )
    )
    await session.execute(stmt)
    await session.commit()

Validating the key: JSONB keys go in as part of the path string. If a customer sends a key with a comma or braces, it breaks the SQL. Validating beforehand is necessary.

A better pattern: validate the allowed keys against a whitelist or with a regex (^[a-z_][a-z0-9_]*$). "Allowing any key" carries risks.

4. Filter posts by metadata content

async def find_posts_with_canonical(session, canonical_url: str) -> list[Post]:
    """
    Useful for detecting duplicates: posts with the same canonical.
    Takes advantage of the GIN.
    """
    stmt = select(Post).where(
        Post.metadata_.contains({"seo": {"canonical": canonical_url}})
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())


async def find_posts_with_custom_field(
    session, key: str, value: Any
) -> list[Post]:
    """
    'Posts where custom_fields.affiliate_id = abc123'.
    """
    stmt = select(Post).where(
        Post.metadata_.contains({"custom_fields": {key: value}})
    )
    result = await session.execute(stmt)
    return list(result.scalars().all())

Both use @> and take advantage of the GIN. If the query is very frequent with a specific field (e.g. metadata.seo.canonical), consider an expression index on top of the GIN:

CREATE INDEX ix_posts_canonical
  ON posts ((metadata #>> '{seo,canonical}'));

It speeds up queries with WHERE metadata #>> '{seo,canonical}' = '...' (an access operator, which GIN doesn't accelerate). The GIN is still there for @> queries.

5. Return validated metadata to the frontend

class PostResponse(BaseModel):
    id: int
    title: str
    slug: str
    body: str
    status: str
    metadata: PostMetadata


@app.get("/posts/{slug}", response_model=PostResponse)
async def get_post(slug: str) -> PostResponse:
    async with SessionLocal() as session:
        post = (
            await session.execute(select(Post).where(Post.slug == slug))
        ).scalar_one_or_none()
        if post is None:
            raise HTTPException(404)
        return PostResponse(
            id=post.id,
            title=post.title,
            slug=post.slug,
            body=post.body,
            status=post.status,
            metadata=PostMetadata.model_validate(post.metadata_),
        )

Pydantic applies the defaults: if the post has no seo, it returns it as null. If it has no custom_fields, it returns {}. The client always receives the full shape.


The anti-pattern smell test

Abuse #1 of "extensible metadata" is turning it into a badly designed mini-database inside a column. Symptoms and fixes:

Symptom 1: a list of sub-entities

# WRONG
post.metadata = {
    "comments": [
        {"id": 1, "user": "...", "body": "...", "created_at": "..."},
        {"id": 2, "user": "...", "body": "...", "created_at": "..."},
        ...  # hundreds
    ]
}

Why: comments need to be listed, counted, paginated, filtered, deleted individually, indexed by user. All of that is trivial with a table, painful with JSONB.

Fix: a comments(id, post_id, user_id, body, created_at) table with an FK to the post.

Symptom 2: aggregations over the content

# WRONG
# "How many posts have 'environment' as a custom_field?"
# You have to walk every post and count.

Why: aggregating requires count(distinct ...) with conditions over variable paths. The SQL gets ugly and doesn't index well.

Fix:

  • If the "custom keys" are finite and known, use typed columns or a fixed set of fields in a Pydantic sub-object.
  • If they're genuinely arbitrary and you need aggregations, a post_custom_fields(post_id, key, value) table. More relational, more indexable.

Symptom 3: lost referential integrity

# WRONG
post.metadata = {"author_id": 42, "category_id": 5}
# If author 42 is deleted, the JSONB is left with an orphan ID.

Why: PostgreSQL doesn't do FKs from inside JSONB. The references dangle.

Fix: foreign keys are typed columns. author_id BIGINT REFERENCES authors(id), not metadata.author_id.

Symptom 4: the field is searched with full-text

# WRONG
post.metadata = {"keywords": "python sqlalchemy postgres jsonb"}
# You want to search "posts that mention sqlalchemy in metadata".

Why: FTS over JSONB is possible but it isn't what the engine is optimized for.

Fix: if your primary use of the field is textual search, a text column + FTS (module 3) wins. JSONB is still useful for structured data, not for long text.

Symptom 5: validation that becomes hell

# WRONG
class PostMetadata(BaseModel):
    seo: PostSEO | None = None
    twitter: TwitterMetadata | None = None
    facebook: FacebookMetadata | None = None
    instagram: InstagramMetadata | None = None
    pinterest: PinterestMetadata | None = None
    linkedin: LinkedInMetadata | None = None
    # ... 30 optional sub-models

Why: if your metadata accumulates 30 sub-objects, it isn't metadata anymore — it's a relational schema in disguise.

Fix: "per-channel metadata" is probably its own entity. A post_channel_metadata(post_id, channel, data JSONB) table with FKs and constraints, where data is JSONB but the shape is decided by the channel.

The quick test

Ask four things when considering adding something to metadata:

  1. Do I need a JOIN with this? If yes, it isn't metadata, it's a related entity.
  2. Do I need complex aggregations (COUNT, SUM, GROUP BY) over this? If yes, probably a column or a table.
  3. Do I need an FK from this to another table? If yes, it's a column.
  4. Am I going to search free text inside this? If yes, a text column + FTS.

If all four are no, JSONB is a good home.


Connection with the integrative project

The "JSONB metadata in posts" component of the guide's final project (module 8) is exactly this pattern. Capsule 08 of this module walks you through the first step (a minimal refactor of the existing Blog API to add metadata with an Alembic migration). The final project extends that by combining it with FTS over the content (module 3), comments partitioning (module 4), and materialized views (module 5).

What you learn here is the foundation. By the time you get to module 8, you'll already know how to declare the column, validate it, migrate data, and query it.


Why does this matter in real work?

1. Custom fields are a universal SaaS requirement. Every customer wants custom fields. Without JSONB metadata, you have two bad options: an entity_custom_fields(entity_id, key, value) table (slow queries, badly indexed) or dummy columns extra1, extra2, ... (a well-known anti-pattern). JSONB solves it elegantly.

2. SEO evolves constantly. Open Graph, Twitter Card, Schema.org, AMP, structured data. Every year there's a new spec. JSONB metadata absorbs it without a migration; columns force an Alembic run on every change.

3. Computed data without an extra table. read_time_minutes, word_count, last_indexed_at. That's three fields. Three new columns is noise. A computed sub-object in metadata is clean and groups what belongs to the same concept.

4. A defensible decision in code review. "Why isn't seo three columns (seo_title, seo_description, seo_canonical)?" Your answer: "Because marketing asked for og_image last month and will ask for twitter_card next month. Each one would be an Alembic migration. In metadata it's a Pydantic shape, no migration, validated at the endpoint."

5. Migrating columns to metadata is a real project. In old projects, the "we add a nullable column every time the business asks" pattern ended with tables of 80 empty columns. Migrating to JSONB is a common refactor. Capsule 08 shows you the Alembic script.


Traps and common mistakes

Mistake 1 (conceptual): mixing customer metadata with computed data, without separating them

Symptom: metadata has read_time_minutes (computed by a job) at the same level as affiliate_id (the customer's custom field). When the customer updates metadata, they overwrite read_time_minutes by accident.

Why it happens: you didn't separate namespaces. Everything goes at the root.

Fix: sub-objects per category. metadata.seo, metadata.computed, metadata.custom_fields. The customer's endpoint only writes to seo and custom_fields. The job only writes to computed. They don't clobber each other.

Mistake 2 (practical): allowing any key in custom_fields

Symptom: a customer passes custom_fields: {"a": "v", "b": "v", ...} with 1000 keys. Your metadata weighs 200 KB.

Why it happens: dict[str, Any] with extra="allow" and no limit.

Fix: validate limits. Field(max_items=20) in Pydantic v2 (or a custom validator). Also: a reasonable length for each key/value (Field(max_length=100) per value). A clear policy: "at most 20 custom fields per post, 100 chars per value."

Mistake 3 (practical): not creating an expression index for frequent queries

Symptom: the query WHERE metadata #>> '{seo,canonical}' = $1 (for detecting duplicates) is slow at 100k posts. EXPLAIN shows a Seq Scan.

Why it happens: a GIN on metadata doesn't accelerate access operators (#>>). You need an expression index.

Fix:

CREATE INDEX ix_posts_canonical
  ON posts ((metadata #>> '{seo,canonical}'));

It's a B-tree on the expression. It speeds up the exact query. The GIN stays for @>.

Mistake 4 (conceptual): assuming metadata replaces the schema

Symptom: everything new goes into metadata. The posts table is left with 4 columns (id, slug, body, metadata) and everything else lives in JSONB. Searching and filtering become inefficient.

Why it happens: "I already have metadata, I'll put everything in it."

Fix: the decision matrix from capsule 05. Status, critical dates, FKs, everything you filter on a lot are columns. Metadata is for what evolves and gets projected more than filtered.

Mistake 5 (practical): the customer sends null where your app expects a string

Symptom: a customer sends seo: {"og_image": null} meaning "don't set the og_image." Your validation accepts it (because og_image: HttpUrl | None). Your frontend renders <img src="null">.

Why it happens: the semantics of "absent" vs "explicitly null" were never defined.

Fix:

  • Clear documentation of the contract.
  • If "absent" and "null" mean the same thing, normalize on save: if og_image is None, don't include the key. model_dump(exclude_none=True) helps.
  • A defensive frontend: never assume a nullable field is a string without checking.

Mistake 6 (conceptual): metadata becomes a history log

Symptom: every field update appends an object to metadata.history = [...]. After a year, history weighs MBs.

Why it happens: metadata gets confused with an audit log.

Fix: history goes in an audit_log table with an FK to the post. Metadata is a snapshot of the current state, not of the history.


Exercises

Exercise 1: classify fields for posts.metadata

Your product team adds a list of fields for posts. For each one, decide: typed column, metadata sub-object (which one), or related table.

a) slug b) published_at c) seo_title d) read_time_minutes (computed) e) tags (each post has 0-5 tags from a fixed set of 30) f) comments (a list of comments) g) featured_image_url h) featured_image_alt i) experiment_variant_for_ab_test (string, only some posts have it) j) version_history (every major change to the post)

See solution

a) slug → typed column with UNIQUE. Critical for URLs, must be unique, indexed.

b) published_atTIMESTAMPTZ column. Filtering and ordering are very frequent.

c) seo_titlemetadata.seo.title. Part of the known SEO set. A strict Pydantic sub-object.

d) read_time_minutesmetadata.computed.read_time_minutes. Computed. The computed sub-object separates it from the customer's fields.

e) tags → a related table (post_tags). You need to: filter posts by tag (frequent), list the most used tags (an aggregation), normalize tag names. A join table with FKs + a tags table.

f) comments → a related table. Long lists, paginated, with an FK to users. JSONB would be an anti-pattern.

g) featured_image_url → a column, or metadata.featured_image.url. If all your posts have a featured image and you use it in listing queries, a column. If it's optional and only rendered on the post's page, it can go in metadata.featured_image.

h) featured_image_alt → the same place as featured_image_url. Glued to the URL: if the URL is a column, featured_image_alt is too. If it's in metadata, metadata.featured_image.alt. Keep together what belongs together.

i) experiment_variant_for_ab_testmetadata.custom_fields.experiment_variant. Only some posts have it, and the experiment's schema changes between tests. A custom field is the place.

j) version_history → a related table (post_versions). Historical, chronological queries, possible diffs. JSONB would be the anti-pattern from symptom 6.

General pattern: what's stable and heavily filtered goes in columns. What's groupable under SEO/branding/features goes in metadata sub-objects. What's historical / a list with FKs goes in tables.

Exercise 2: implement a partial update without clobbering other fields

Write a function update_seo(session, post_id, patch: PostSEO_Patch) where PostSEO_Patch has all the SEO fields optional. The function must:

  1. Update only the fields the patch has defined (not None).
  2. Not touch metadata.computed or metadata.custom_fields.
  3. If SEO didn't exist before and the patch has the required fields to build a complete SEO, create it.
  4. If after the merge the SEO is incomplete, fail with a clear error.
See solution
from pydantic import BaseModel, ValidationError, Field, HttpUrl
from sqlalchemy import select
from fastapi import HTTPException


class PostSEO(BaseModel):
    title: str = Field(min_length=1, max_length=160)
    description: str = Field(min_length=1, max_length=320)
    canonical: HttpUrl
    og_image: HttpUrl | None = None
    twitter_card: str | None = Field(default=None, max_length=50)


class PostSEO_Patch(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=160)
    description: str | None = Field(default=None, min_length=1, max_length=320)
    canonical: HttpUrl | None = None
    og_image: HttpUrl | None = None
    twitter_card: str | None = Field(default=None, max_length=50)


async def update_seo(session, post_id: int, patch: PostSEO_Patch) -> PostSEO:
    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")

    # Load the current SEO (may be None or a dict)
    current_seo_dict = post.metadata_.get("seo") or {}

    # Apply the patch (only non-None fields)
    patch_dict = patch.model_dump(exclude_none=True, mode="json")
    merged = {**current_seo_dict, **patch_dict}

    # Validate that the merge is a complete PostSEO
    try:
        merged_seo = PostSEO.model_validate(merged)
    except ValidationError as exc:
        raise HTTPException(
            422,
            f"Incomplete SEO after the patch: {exc.errors()}",
        )

    # Reassign the seo sub-object (MutableDict detects the root change)
    post.metadata_["seo"] = merged_seo.model_dump(mode="json")
    await session.commit()
    return merged_seo

Why it works:

  • exclude_none=True filters the patch's Nones — it doesn't overwrite existing fields with None.
  • The merge {**current, **patch} applies only what came in.
  • Re-validating against the full PostSEO guarantees that after the patch, the shape is valid.
  • Reassigning post.metadata_["seo"] (rather than mutating the sub-object in place) is what MutableDict needs in order to detect the change.
  • metadata_["computed"] and metadata_["custom_fields"] are left untouched.

A more atomic alternative (without loading the post):

It's more complex because you'd need to read the current seo from SQL to merge it. A loaded MutableDict works well here; func.jsonb_set requires chaining several calls (one per patch field).

Exercise 3: a duplicate-detection query

Your SEO team asks you for an endpoint that returns posts with the same canonical URL (a typical problem). Write the SQLAlchemy query and the SQL it generates. Consider performance: what index do you propose?

See solution
from sqlalchemy import select, func


async def find_duplicate_canonicals(session) -> list[dict[str, Any]]:
    """
    Returns [{"canonical": "...", "count": N, "post_ids": [...]}, ...]
    for canonicals that appear in more than one post.
    """
    canonical_expr = Post.metadata_["seo"]["canonical"].astext.label("canonical")
    stmt = (
        select(
            canonical_expr,
            func.count(Post.id).label("count"),
            func.array_agg(Post.id).label("post_ids"),
        )
        .where(Post.metadata_["seo"]["canonical"].astext.is_not(None))
        .group_by(canonical_expr)
        .having(func.count(Post.id) > 1)
        .order_by(func.count(Post.id).desc())
    )
    result = await session.execute(stmt)
    return [
        {"canonical": r.canonical, "count": r.count, "post_ids": list(r.post_ids)}
        for r in result.all()
    ]

Generated SQL:

SELECT
  (posts.metadata #>> '{seo,canonical}') AS canonical,
  count(posts.id) AS count,
  array_agg(posts.id) AS post_ids
FROM posts
WHERE (posts.metadata #>> '{seo,canonical}') IS NOT NULL
GROUP BY (posts.metadata #>> '{seo,canonical}')
HAVING count(posts.id) > 1
ORDER BY count(posts.id) DESC;

Proposed index: an expression index on the full path:

CREATE INDEX ix_posts_canonical
  ON posts ((metadata #>> '{seo,canonical}'))
  WHERE metadata #>> '{seo,canonical}' IS NOT NULL;

Why this index:

  • It's a B-tree on the exact expression the query uses. PostgreSQL will use it for the GROUP BY.
  • WHERE ... IS NOT NULL makes it a partial index: it only indexes rows where the canonical exists. Smaller, faster.
  • The GIN on metadata (which you already have) doesn't accelerate this query — it's an access operator (#>>), not a search. Without the expression index, the query would be a Seq Scan.

Validation:

EXPLAIN ANALYZE
SELECT (metadata #>> '{seo,canonical}') AS canonical, count(*)
FROM posts
WHERE (metadata #>> '{seo,canonical}') IS NOT NULL
GROUP BY canonical
HAVING count(*) > 1;
-- Before the index: Seq Scan + Sort + Group
-- After the index: Index Scan + Group (much faster past 100k posts)

Exercise 4: spot the anti-pattern

Review this code and apply the smell test. What things should come out of metadata, and where should they go?

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    customer_id: Mapped[int]
    metadata_: Mapped[dict] = mapped_column("metadata", JSONB, default=dict)
    # metadata = {
    #   "items": [
    #     {"product_id": 12, "qty": 2, "price": 19.99},
    #     {"product_id": 45, "qty": 1, "price": 49.99},
    #   ],
    #   "shipping_address": {"street": "...", "city": "...", "country": "MX"},
    #   "payment": {"method": "card", "last4": "1234", "status": "paid"},
    #   "applied_coupons": [{"code": "SUMMER20", "discount": 10.00}],
    #   "internal_notes": [
    #     {"by": "agent_5", "text": "...", "ts": "2026-04-01T..."},
    #     ... (can grow)
    #   ],
    #   "tracking_events": [
    #     {"status": "shipped", "ts": "..."},
    #     {"status": "in_transit", "ts": "..."},
    #     ... (grows with every courier update)
    #   ]
    # }
See solution

What's wrong and where it goes:

  1. items → an order_items table.

    • Smell test: you need to list items, compute a total, add/remove individual items, join with products. Yes, yes, yes, yes.
    • Schema: order_items(id, order_id, product_id, qty, price). FK to the product, FK to the order.
  2. shipping_address → ambiguous.

    • If the address is immutable after checkout and only read alongside the order, JSONB is acceptable (a snapshot of the moment of purchase).
    • If you need to filter orders by country/city ("how many orders to Mexico this month?"), then expression indexes on metadata #>> '{shipping_address,country}' or dedicated columns.
    • Decision: it depends on the usage. For an MVP, JSONB. For production with analytics, an extra shipping_country column.
  3. payment → mixed.

    • method, last4, status are typed columns: status is filtered on a lot, method gets reported on, last4 is always displayed. Plus some optional ones like processor_response that do go in JSONB.
    • Refactor: columns payment_method, payment_status, payment_last4 + payment_metadata JSONB for the details.
  4. applied_coupons → an order_coupons table.

    • You need to: report coupon usage (which coupon how many people used), validate that a coupon isn't applied twice to the same order, add/remove coupons. A table.
  5. internal_notes → an order_notes table.

    • They grow without limit, authorship with an FK to users, chronological ordering, possibly indexed. JSONB would be a disaster over time.
  6. tracking_events → a tracking_events table.

    • Growing history. Every courier update appends an entry. JSONB becomes an infinite log. A table with an FK to the order and a status enum.

Refactored schema:

class Order(Base):
    id: ...
    customer_id: ... ForeignKey("customers.id")
    payment_method: Mapped[str]
    payment_status: Mapped[str]
    payment_last4: Mapped[str | None]
    shipping_country: Mapped[str]   # extracted for queries
    metadata_: Mapped[dict] = mapped_column("metadata", JSONB, default=dict)
    # metadata = only what is genuinely extensible:
    #   - shipping_address (snapshot, optional info like delivery instructions)
    #   - payment_metadata (processor details)
    #   - custom_fields (if the customer wants extras)


class OrderItem(Base):
    order_id: ...
    product_id: ...
    qty: ...
    price: ...


class OrderCoupon(Base):
    order_id: ...
    coupon_code: ...
    discount: ...


class OrderNote(Base):
    order_id: ...
    by_user_id: ...
    text: ...
    created_at: ...


class TrackingEvent(Base):
    order_id: ...
    status: ...
    created_at: ...

Lesson: the original metadata was a textbook anti-pattern. Every list was a relational entity in disguise. The final JSONB ends up small and focused on what's genuinely extensible.

Exercise 5: namespacing by category

Design the Pydantic schema for users.metadata in a SaaS where the user has:

  • UI preferences: theme (dark/light/auto), language, layout.
  • Notifications: opt-in per channel (email/slack/sms) and per type (alerts/digest/marketing).
  • Data computed by a job: last login, session count in the last month, engagement score.
  • Custom fields from the tenant's admin: arbitrary keys from the set the tenant allows.
See solution
from typing import Literal
from datetime import datetime
from pydantic import BaseModel, Field


class UIPreferences(BaseModel):
    theme: Literal["dark", "light", "auto"] = "auto"
    language: str = Field(default="es", min_length=2, max_length=5)
    layout: Literal["compact", "comfortable"] = "comfortable"


class NotificationChannelPrefs(BaseModel):
    alerts: bool = True
    digest: bool = False
    marketing: bool = False


class NotificationPreferences(BaseModel):
    email: NotificationChannelPrefs = NotificationChannelPrefs()
    slack: NotificationChannelPrefs = NotificationChannelPrefs(
        alerts=False, digest=False, marketing=False
    )
    sms: NotificationChannelPrefs = NotificationChannelPrefs(
        alerts=False, digest=False, marketing=False
    )


class UserComputed(BaseModel):
    last_login_at: datetime | None = None
    sessions_last_30d: int = Field(ge=0, default=0)
    engagement_score: float = Field(ge=0, le=100, default=0)


class UserMetadata(BaseModel):
    ui: UIPreferences = UIPreferences()
    notifications: NotificationPreferences = NotificationPreferences()
    computed: UserComputed = UserComputed()
    custom_fields: dict[str, str | int | bool | None] = Field(default_factory=dict)
    model_config = {"extra": "allow"}

Decisions explained:

  • Sub-models per category give you namespacing. The endpoint that updates ui can't touch notifications by accident.
  • Literal on theme and layout rejects unexpected values. Without it, a client could set theme: "rainbow" and break the frontend.
  • NotificationChannelPrefs is reused for all three channels. DRY and consistent.
  • computed has its own namespace so the job doesn't clobber the user's data.
  • custom_fields accepts arbitrary keys but typed values (not arbitrary nested objects — it limits complexity).
  • extra="allow" at the top level for future fields that don't have a schema yet.

How it's used in endpoints:

# PATCH /users/{id}/preferences/ui
class UIPatch(BaseModel):
    theme: Literal["dark", "light", "auto"] | None = None
    language: str | None = None
    layout: Literal["compact", "comfortable"] | None = None


@app.patch("/users/{user_id}/preferences/ui")
async def patch_ui(user_id: int, patch: UIPatch) -> UIPreferences:
    # Merge with the current UI, validate, assign to user.metadata_["ui"]
    ...

Each category has its endpoint. Each endpoint only touches its sub-object. Clean, maintainable.


Summary and next step

In this capsule you learned the second canonical JSONB pattern:

  • Extensible metadata is per-instance information that evolves with the customer and the use cases (vs configuration, which is per-entity and controlled by your team).
  • Three levels of strictness inside the same JSONB: known fields (strict Pydantic), custom_fields (a dict with typed values), extra="allow" (future keys).
  • Namespacing with sub-objects (metadata.seo, metadata.computed, metadata.custom_fields) keeps different actors from clobbering each other's data.
  • Partial updates with func.jsonb_set, or reassigning a sub-object with MutableDict, depending on the concurrency.
  • Expression indexes complement the GIN for queries with access operators (#>>).
  • The anti-pattern smell test: do you need JOINs, complex aggregations, FKs, or textual search? If yes to any of them, it isn't metadata, it's a relational entity.

Before moving on you should be able to:

  • Design Pydantic metadata with layered validation (strict + custom + extras).
  • Decide whether a new field goes in metadata or another structure by applying the smell test.
  • Implement partial updates that don't clobber each other across actors.
  • Identify metadata that's "become a badly designed mini-DB" and propose a refactor.

Next capsule — Pattern: polymorphic data. The third and final canonical pattern. When your table stores entities of the same logical type but with a different shape depending on a discriminating key (event_type, notification_type, audit_action), JSONB with Pydantic discriminated unions is the right pattern. You'll go deeper on what you skimmed in capsule 04, now with queries by discriminator, the GIN put to work, and the anchor case of events/audit_log that shows up in real systems.


Resources

  1. PostgreSQL Wiki — JSONB — the community discussion on usage patterns.
  2. Crunchy Data — Modeling with JSONB — an applied analysis of when it's worth it.
  3. Pydantic v2 — Models with extra fields — the reference for extra="allow" / "forbid" / "ignore".
  4. PostgreSQL 16 — Expression indexes — the official reference.
  5. PostgreSQL 16 — Partial indexes — for conditional expression indexes.
  6. Open Graph Protocol — the Open Graph reference, to understand why SEO evolves.
  7. SQLAlchemy 2.0 — func.array_agg — used in the duplicate-detection exercise.

Module 2 — Advanced PostgreSQL for Backend Guide

Next capsule: Pattern: polymorphic data — events, audit logs, and discriminator queries with Pydantic discriminated unions.