Module 2: JSONB with SQLAlchemy and usage patterns

Module project: extend the Blog API with `posts.metadata` JSONB

Next →

What are you going to build and why?

You're going to refactor the Blog API you built in guide #8 (PostgreSQL & SQLAlchemy) to replace several loose optional columns with a metadata JSONB column validated by Pydantic and indexed with GIN. The delivery is a PR-style change on your Blog API repo: a three-phase Alembic migration (zero-downtime), updated SQLAlchemy/FastAPI code, tests for the happy path and the edge cases, and before/after benchmarks of the queries that benefited most.

This project integrates everything from the module: JSONB column declaration (capsule 02), MutableDict and func.jsonb_set for mutations (capsule 03), Pydantic validation with sub-models and extensibility (capsule 04), the extensible metadata pattern (capsule 06), and GIN-indexed queries from module 1.

It's also the first step of the integrative project in module 8 (the end of the complete guide). There you'll extend this refactor by combining it with FTS (module 3), partitioning of comments (module 4), and materialized views (module 5). Doing this project well saves you work later on.

Project objective

By completing this project:

  • You'll design the posts.metadata schema with Pydantic (seo, computed, custom_fields sub-objects).
  • You'll migrate from the old schema (seo_title VARCHAR, seo_description VARCHAR, seo_canonical VARCHAR, read_time_minutes INT columns) to the new one (metadata JSONB) without losing data and with a working rollback.
  • You'll adapt the existing endpoints (POST /posts, GET /posts/{slug}, PATCH /posts/{id}) to validate metadata with Pydantic.
  • You'll create two new endpoints: GET /posts/duplicates-by-canonical (detects SEO duplicates) and PATCH /posts/{id}/seo (partial update with func.jsonb_set).
  • You'll index with GIN over metadata and an expression index over metadata #>> '{seo,canonical}'.
  • You'll measure the before/after with EXPLAIN ANALYZE on at least two queries.
  • You'll document the decisions in ARCHITECTURE.md.

How it fits with what you learned

Module conceptWhere it's used in the project
Capsule 02: Mapped[dict], accessorsDeclaring metadata, queries with Post.metadata_["seo"]["canonical"].astext
Capsule 03: MutableDict, func.jsonb_setPartial updates in PATCH /posts/{id}/seo
Capsule 04: Pydantic validation with extra="allow"The PostMetadata schema with sub-models and custom_fields
Capsule 05: decision matrixJustifying why metadata and not columns
Capsule 06: namespacing by sub-objectsThe seo/computed/custom_fields structure
GIN from module 1, capsule 05Index over metadata for @>
Expression indexes from module 1, capsule 05Index over metadata #>> '{seo,canonical}'
Zero-downtime Alembic migrations (guide #13, module 4)The multi-phase pattern for not breaking the app in production

Think of the project as an integration test of everything you studied. If you deliver it with the seven checkpoints, you know the module is consolidated.

Technical specifications

Stack

  • Language: Python 3.11+
  • Main framework: FastAPI 0.110+
  • ORM: async SQLAlchemy 2.0+ with asyncpg
  • Validation: Pydantic 2.6+
  • Migrations: Alembic 1.13+
  • Tests: pytest 8+ with pytest-asyncio
  • DB: PostgreSQL 16+

Initial setup

Assume you already have the Blog API from guide #8 running. If you don't, clone the reference repo (any working mini Blog API with FastAPI + async SQLAlchemy 2.0 works). Create a feature branch:

cd /path/to/your/blog-api
git checkout -b feature/posts-metadata-jsonb
pip install --upgrade "sqlalchemy[asyncio]>=2.0" "asyncpg>=0.29" "pydantic>=2.6" "alembic>=1.13"

Assumed initial state

Your Blog API has a Post model roughly like this:

# app/models.py — CURRENT STATE (before the refactor)
from datetime import datetime
from typing import Literal

from sqlalchemy import BigInteger, String, Text
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]

    # Optional fields we're going to migrate to metadata JSONB
    seo_title: Mapped[str | None] = mapped_column(String(160))
    seo_description: Mapped[str | None] = mapped_column(String(320))
    seo_canonical: Mapped[str | None] = mapped_column(String(500))
    read_time_minutes: Mapped[int | None] = mapped_column(default=0)

If your fields are slightly different, adapt the names but keep the spirit: there are 3-5 optional columns with data you would naturally group under "metadata".

Required features

1. Pydantic metadata schema

In app/schemas/post_metadata.py:

from datetime import datetime
from pydantic import BaseModel, Field, HttpUrl


class PostSEO(BaseModel):
    """Strict schema: required fields when seo is present."""
    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 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):
    """Complete metadata structure."""
    seo: PostSEO | None = None
    computed: PostComputed | None = None
    custom_fields: dict[str, str | int | bool | None] = Field(default_factory=dict)
    model_config = {"extra": "allow"}

Constraint: sub-objects by category. Do NOT use a flat dict.

2. Updated SQLAlchemy model

In app/models.py — FINAL state after the refactor:

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]

    # New JSONB column
    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"),
    )

Constraints:

  • default=dict (not {}).
  • nullable=False with a server_default in the migration.
  • GIN over metadata declared in __table_args__.
  • The old columns (seo_title, etc.) are no longer in the model after phase 3 of the migration.

3. Three-phase Alembic migration (zero-downtime)

In alembic/versions/. Each phase is a separate migration.

Phase A — Add the metadata column with a default and backfill

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


revision = "add_metadata_jsonb"
down_revision = "<the previous revision>"


def upgrade() -> None:
    # Add the column as nullable=False with server_default '{}'
    op.add_column(
        "posts",
        sa.Column(
            "metadata",
            postgresql.JSONB(astext_type=sa.Text()),
            nullable=False,
            server_default=sa.text("'{}'::jsonb"),
        ),
    )

    # Create the GIN index
    op.create_index(
        "ix_posts_metadata_gin",
        "posts",
        ["metadata"],
        postgresql_using="gin",
    )


def downgrade() -> None:
    op.drop_index("ix_posts_metadata_gin", table_name="posts")
    op.drop_column("posts", "metadata")

After running this migration, all rows have metadata = {}. The app keeps reading and writing the old columns (seo_title, etc.) without breaking anything.

Phase B — Backfill of the existing data

# alembic/versions/202605xx_backfill_metadata_from_columns.py
import sqlalchemy as sa
from alembic import op


revision = "backfill_metadata"
down_revision = "add_metadata_jsonb"


def upgrade() -> None:
    """
    For each post, build metadata.seo and metadata.computed
    from the old columns. Idempotent: if metadata already
    has the data, it doesn't overwrite it.
    """
    op.execute("""
        UPDATE posts
        SET metadata = jsonb_build_object(
          'seo', CASE
              WHEN seo_title IS NOT NULL OR seo_description IS NOT NULL
                   OR seo_canonical IS NOT NULL
              THEN jsonb_build_object(
                  'title', seo_title,
                  'description', seo_description,
                  'canonical', seo_canonical
              )
              ELSE NULL
          END,
          'computed', CASE
              WHEN read_time_minutes IS NOT NULL THEN
                  jsonb_build_object(
                      'read_time_minutes', read_time_minutes,
                      'word_count', 0
                  )
              ELSE NULL
          END
        )
        WHERE metadata = '{}'::jsonb;
    """)


def downgrade() -> None:
    """
    Revert: clears metadata. The old columns still hold
    the original values.
    """
    op.execute("UPDATE posts SET metadata = '{}'::jsonb;")

Details:

  • WHERE metadata = '{}'::jsonb makes the migration idempotent: if it already ran, it doesn't overwrite.
  • CASE WHEN ... THEN ... ELSE NULL END avoids empty sub-objects when there was no data.
  • If your production has millions of posts, this UPDATE locks rows: instead of a single UPDATE, run the backfill in batches (LIMIT + offset by id) from a Python script. For your Blog API locally with thousands of posts, a single UPDATE is fine.

Phase C — Drop the old columns (after validating)

# alembic/versions/202605xx_drop_seo_columns.py
import sqlalchemy as sa
from alembic import op


revision = "drop_seo_columns"
down_revision = "backfill_metadata"


def upgrade() -> None:
    op.drop_column("posts", "seo_title")
    op.drop_column("posts", "seo_description")
    op.drop_column("posts", "seo_canonical")
    op.drop_column("posts", "read_time_minutes")


def downgrade() -> None:
    op.add_column("posts", sa.Column("seo_title", sa.String(160)))
    op.add_column("posts", sa.Column("seo_description", sa.String(320)))
    op.add_column("posts", sa.Column("seo_canonical", sa.String(500)))
    op.add_column("posts", sa.Column("read_time_minutes", sa.Integer))
    # Backfill from metadata for a complete downgrade
    op.execute("""
        UPDATE posts
        SET
          seo_title = metadata #>> '{seo,title}',
          seo_description = metadata #>> '{seo,description}',
          seo_canonical = metadata #>> '{seo,canonical}',
          read_time_minutes = (metadata #>> '{computed,read_time_minutes}')::int;
    """)

Critical constraint: do NOT run Phase C in production until the deploy with the new code is stable. The pattern is:

  1. Phase A in production.
  2. Phase B in production.
  3. Deploy of the new code (which reads/writes metadata and no longer touches the old columns).
  4. Validate for 24-48h.
  5. Phase C.

In your local environment you can run all three back to back, but document the pattern in ARCHITECTURE.md.

4. Updated endpoints

POST /posts

Receives a CreatePostRequest that includes an optional metadata: PostMetadata.

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.post("/posts", status_code=201)
async def create_post(payload: CreatePostRequest, session: AsyncSession = Depends(get_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}

GET /posts/{slug}

Returns metadata validated with Pydantic.

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, session: AsyncSession = Depends(get_session)):
    post = (await session.execute(select(Post).where(Post.slug == slug))).scalar_one_or_none()
    if post is None:
        raise HTTPException(404, "post not found")
    return PostResponse(
        id=post.id,
        title=post.title,
        slug=post.slug,
        body=post.body,
        status=post.status,
        metadata=PostMetadata.model_validate(post.metadata_),
    )

PATCH /posts/{id}/seo — new endpoint

Partial update using func.jsonb_set. Receives a PostSEOPatch (all fields optional).

class PostSEOPatch(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


@app.patch("/posts/{post_id}/seo", response_model=PostSEO)
async def patch_post_seo(
    post_id: int,
    patch: PostSEOPatch,
    session: AsyncSession = Depends(get_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")

    current_seo = post.metadata_.get("seo") or {}
    patch_dict = patch.model_dump(exclude_none=True, mode="json")
    merged = {**current_seo, **patch_dict}

    try:
        merged_seo = PostSEO.model_validate(merged)
    except ValidationError as exc:
        raise HTTPException(422, f"Incomplete SEO after patch: {exc.errors()}")

    post.metadata_["seo"] = merged_seo.model_dump(mode="json")
    await session.commit()
    return merged_seo

GET /posts/duplicates-by-canonical — new endpoint

Detects posts with the same canonical. Uses the proposed expression index.

class DuplicateGroup(BaseModel):
    canonical: str
    count: int
    post_ids: list[int]


@app.get("/posts/duplicates-by-canonical", response_model=list[DuplicateGroup])
async def list_duplicates(session: AsyncSession = Depends(get_session)):
    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 [
        DuplicateGroup(canonical=r.canonical, count=r.count, post_ids=list(r.post_ids))
        for r in result.all()
    ]

5. Additional indexes

Create them with a SQL script (or as part of phase A if you prefer):

-- Expression index for the duplicates query
CREATE INDEX ix_posts_canonical
  ON posts ((metadata #>> '{seo,canonical}'))
  WHERE metadata #>> '{seo,canonical}' IS NOT NULL;

It's a partial index — it only indexes posts with canonical defined. Smaller, faster.

6. Tests

In tests/test_metadata.py. Use pytest-asyncio.

Required tests:

a) Create a post with valid metadata → 201, metadata persisted, returned validated. b) Create a post with incomplete SEO (missing description) → 422. c) A GET of a post returns metadata with the Pydantic shape (defaults applied). d) PATCH SEO with a partial patch (only title) → SEO updated, other fields preserved. e) PATCH SEO that would leave the SEO incomplete → 422. f) Duplicate detection works: create two posts with the same canonical and verify they show up grouped. g) Persistence test after reopening the session (verifies that MutableDict works): mutate post.metadata_["seo"] = {...}, commit, open a new session, verify persistence.

7. Before/after benchmark

In BENCHMARKS.md. Measure at least two queries with EXPLAIN ANALYZE:

a) Search by canonical (before: WHERE seo_canonical = '...' with a B-tree index; after: WHERE metadata #>> '{seo,canonical}' = '...' with an expression index).

b) Duplicate detection (before: GROUP BY seo_canonical with a B-tree index; after: GROUP BY (metadata #>> '{seo,canonical}') with an expression index).

c) (Optional) Containment filtering that wasn't possible before (e.g. "posts with any SEO field defined": before it was WHERE seo_title IS NOT NULL OR seo_description IS NOT NULL OR ...; after it's WHERE metadata ? 'seo').

For the benchmark, make sure to:

  • Have at least 10k posts with synthetic data (seeding script).
  • Run each query 5 times after a warmup.
  • Report the median of the percentiles (not the average).
  • Document the environment (PostgreSQL version, hardware, dataset size).

Validations and error handling

What must be validated

  • slug matches the regex ^[a-z0-9-]+$ (already in the old schema, keep it).
  • metadata.seo, if present, has the 3 required fields (title, description, canonical).
  • metadata.seo.canonical is a valid URL.
  • metadata.seo.title length 1-160; description 1-320.
  • metadata.computed.word_count and read_time_minutes are >= 0.
  • custom_fields has values of type str | int | bool | None (not arbitrary nested objects).

Errors that must be handled

  • Nonexistent post (GET /posts/{slug}, PATCH /posts/{id}/seo): 404.
  • Pydantic validation fails: 422 with detail on which field and why.
  • Duplicate slug on POST: 409 Conflict (UNIQUE constraint).
  • A SEO patch that leaves the SEO incomplete: 422 with a clear message.

Minimal implementation example

A skeleton the student must extend. It is NOT the complete solution.

# app/main.py — minimal skeleton
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.api.posts import router as posts_router


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


app = FastAPI(title="Blog API with JSONB metadata")
app.include_router(posts_router, prefix="/posts", tags=["posts"])


@app.get("/health")
async def health():
    return {"status": "ok"}
# app/api/posts.py — skeleton
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from app.deps import get_session

router = APIRouter()


@router.post("", status_code=201)
async def create_post(...):
    raise NotImplementedError


@router.get("/{slug}")
async def get_post(slug: str, session: AsyncSession = Depends(get_session)):
    raise NotImplementedError


@router.patch("/{post_id}/seo")
async def patch_seo(...):
    raise NotImplementedError


@router.get("/duplicates-by-canonical")
async def list_duplicates(...):
    raise NotImplementedError

To run it:

alembic upgrade head
uvicorn app.main:app --reload

Evaluation rubric (self-check)

Total: 100 points. Passing: ≥70 points.

Functionality (45 points)

  • (5) Post model updated with metadata_ JSONB + MutableDict + GIN.
  • (5) Pydantic PostMetadata schema with seo, computed, custom_fields sub-objects and extra="allow".
  • (10) Alembic migration Phase A (add column + GIN) runs cleanly and is reversible.
  • (10) Migration Phase B (backfill) is idempotent; runs twice without overwriting data.
  • (5) Migration Phase C (drop old columns) runs cleanly and has a downgrade that restores from metadata.
  • (5) The POST /posts endpoint validates metadata with Pydantic; incomplete SEO → 422.
  • (5) The PATCH /posts/{id}/seo endpoint applies a partial patch using func.jsonb_set or MutableDict.
  • (5) The GET /posts/duplicates-by-canonical endpoint returns groups of posts with the same canonical.

Code quality (25 points)

  • (5) Complete type hints: Mapped[dict[str, Any]], not dict.
  • (5) default=dict, not default={}.
  • (5) Pydantic validation on read too, not only on write.
  • (5) model_dump(mode="json") when persisting (no serialization errors show up).
  • (5) Async endpoints, session injected via Depends, no global engines in handlers.

Indexing and performance (15 points)

  • (5) GIN over metadata declared in __table_args__ and created by the migration.
  • (5) Expression index over metadata #>> '{seo,canonical}' created to speed up duplicate detection.
  • (5) EXPLAIN ANALYZE in BENCHMARKS.md shows a Bitmap Index Scan or Index Scan (not a Seq Scan) for the optimized queries.

Tests (10 points)

  • (3) Happy path tests for POST, GET, PATCH.
  • (3) Validation test verifying that incomplete SEO → 422.
  • (2) Duplicates test that creates two posts with the same canonical and validates the response.
  • (2) Persistence test after a new session (confirms that MutableDict works).

Documentation (5 points)

  • (2) ARCHITECTURE.md documents why JSONB for metadata (not columns, not a key/value table).
  • (1) ARCHITECTURE.md documents the three-phase migration pattern.
  • (1) BENCHMARKS.md with an explicit methodology (warmup, runs, percentiles).
  • (1) BENCHMARKS.md with before/after results for at least two queries.

Extra credit (optional, up to +10 points)

  • (+3) Backfill in batches with a Python script (not a single UPDATE) for large datasets.
  • (+2) Synthetic data generation script (10k+ posts with varied metadata) for benchmarks.
  • (+2) A GET /posts/by-custom-field?key=X&value=Y endpoint that filters using contains and takes advantage of GIN.
  • (+3) Concurrency test: two simultaneous requests to PATCH /posts/{id}/seo don't overwrite each other (use func.jsonb_set with different paths).

Common mistakes in this project

Mistake 1: forgetting MutableDict and debugging for hours

Symptom: the test is green, but when you reopen the session the change isn't there. Or in production, the PATCHes seem not to persist.

Why it happens: you declared Mapped[dict[str, Any]] with plain JSONB without MutableDict.as_mutable(JSONB).

How to fix it: always wrap with MutableDict.as_mutable(JSONB) when you're going to mutate the dict in-place.

Mistake 2: backfill with a missing WHERE, not idempotent

Symptom: you run migration B twice and the second one runs the UPDATE again, potentially overwriting data a user modified via the API between Phase B and Phase C.

Why it happens: you forgot WHERE metadata = '{}'::jsonb or an equivalent.

How to fix it: every backfill must be idempotent. WHERE metadata = '{}'::jsonb (only populate if empty) or WHERE NOT (metadata ? 'seo') (only if seo doesn't exist).

Mistake 3: dropping columns in Phase C before deploying the new code

Symptom: after Phase C, the servers still running the old code try to SELECT/UPDATE the removed columns and crash.

Why it happens: you ran Phase C too soon.

How to fix it: the pattern is Phase A → Phase B → deploy of the new code → validate → Phase C. Document this in ARCHITECTURE.md even if your local environment doesn't require the time split.

Mistake 4: a PATCH that overwrites all the metadata

Symptom: PATCH /seo erases metadata.computed or metadata.custom_fields that were populated.

Why it happens: you assigned post.metadata_ = patch_dict instead of post.metadata_["seo"] = merged_seo_dict.

How to fix it: modify only the corresponding sub-key. For func.jsonb_set, the path is '{seo}', not the whole column.

Mistake 5: validating on write but not on read

Symptom: malformed data gets in through a manual migration or a direct script. When a user queries the post, the frontend receives inconsistent data.

Why it happens: you trust that the database always has valid data.

How to fix it: PostMetadata.model_validate(post.metadata_) before returning. If it fails, 500 with a clear cause and a log so ops can investigate.

Mistake 6: forgetting the expression index, slow duplicates queries

Symptom: GET /posts/duplicates-by-canonical with 100k posts takes seconds.

Why it happens: GIN over metadata doesn't speed up access operators (#>>). You need an expression index.

How to fix it: CREATE INDEX ON posts ((metadata #>> '{seo,canonical}')). Confirm with EXPLAIN that it's used.

Mistake 7: tests that pass in the same session (they don't catch the mutation bug)

Symptom: all tests green in CI, the bug shows up in production.

Why it happens: the tests verify mutation + read in the same session. SQLAlchemy returns the dict mutated in memory without persisting it.

How to fix it: every test that verifies persistence must open a new session to validate. Use a fixture that provides a reusable session_factory.

What to do if you get stuck?

  • If migration A fails: read the complete error. The most common one is that a column called metadata already exists (DeclarativeBase uses it internally). Rename it to _metadata or keep it as metadata with a Python-side rename, whichever you prefer.

  • If the backfill doesn't fill in the expected data: run SELECT id, metadata FROM posts LIMIT 10; after Phase B. Check the shape. If everything is {}, review the migration's WHERE (it's probably filtering everything out).

  • If GIN isn't used in the query: use EXPLAIN ANALYZE to check. If it's a Seq Scan, make sure that: (a) the query uses @> or ? (search operators), (b) the index exists (\d posts in psql), (c) the table has enough rows for the planner to choose the index (with <100 rows, the planner prefers a Seq Scan; add more data to the benchmark).

  • If the persistence tests fail when reopening the session: the first suspect is always MutableDict. Confirm that the column is declared as MutableDict.as_mutable(JSONB).

  • If the PATCH endpoint responds 500 instead of 422 on bad validation: you're catching Exception instead of ValidationError specifically, or you aren't wrapping the validation in a try/except.

Resources for the project

  1. SQLAlchemy 2.0 — JSONB and MutableDict — the official reference.
  2. Pydantic v2 — Models and validation — for the schemas.
  3. Alembic — Migration patterns — common migration patterns.
  4. PostgreSQL 16 — jsonb_build_object — used in the backfill.
  5. PostgreSQL 16 — Expression indexes — to speed up duplicate detection.
  6. FastAPI — Dependencies — to inject the session.
  7. pytest-asyncio — for async tests.

What comes next

What you built here is the direct foundation of the integrative project in module 8 (the close of the complete guide). There:

  • You'll extend posts.metadata with a generated tsvector field for Full-Text Search in Spanish (module 3).
  • You'll partition the comments table by month (module 4).
  • You'll create a top_posts_weekly materialized view that uses queries combining metadata + aggregations (module 5).
  • You'll coordinate an FTS re-indexing job with pg_advisory_lock (module 6).
  • You'll implement recursive categories with WITH RECURSIVE (module 8).

Before moving on to module 3, make sure that:

  • Your project PR has the key files committed: updated code, Alembic migrations, BENCHMARKS.md, ARCHITECTURE.md.
  • The tests pass (pytest -v).
  • alembic upgrade head runs cleanly on a fresh database.
  • alembic downgrade -3 also runs cleanly (reverts the three phases).
  • The rubric honestly gives you ≥70 points.

If you deliver with less than 70, identify what you're missing and complete it before moving on. What you leave half-done here is going to cost you double in module 8.

Module 2 — Advanced PostgreSQL for Backend Guide

You built it: a Blog API with posts.metadata JSONB validated by Pydantic, indexed with GIN, migrated in three zero-downtime phases, and tested. You have the first pillar of the final integrative project ready.

Next module: Full-Text Search + pg_trgm. Your Blog API needs free-text search over title and body with relevance and typo tolerance. PostgreSQL FTS solves it without adding Elasticsearch to the stack.

← PreviousPattern: polymorphic data with JSONB and discriminated unionsNext →Module 3: Full-Text Search + pg_trgm — a quality search engine without Elasticsearch