Final project — phase 1: setup, baseline, 3 components (JSONB, FTS, MV)

The integrating project begins. This capsule covers phase 1: setting up the project, measuring the baseline before any change, and applying the first 3 components (JSONB metadata, FTS, materialized view). Capsule 07 covers phase 2 (partitioning, advisory lock, recursive categories). Capsule 08 closes with documentation.

This is real work — the capsule lays out the plan, the implementation takes 3-4 hours on your own time.

---

Project setup

Option A: use the Blog API from guide #8

If you completed guide #8 (PostgreSQL & SQLAlchemy from the Backend Python path), you have a working Blog API. Clone/copy it to a new branch for the refactor:

git clone <your-blog-api-repo> blog-api-advanced
cd blog-api-advanced
git checkout -b advanced-pg-refactor

Option B: minimal scaffold

If you don't have the Blog API, a minimal schema:

# app/models/__init__.py
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(200), unique=True)
    name: Mapped[str]


class Category(Base):
    __tablename__ = "categories"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    parent_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"), nullable=True)


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    category_id: Mapped[int] = mapped_column(ForeignKey("categories.id"))
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(Text)
    published_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
    view_count: Mapped[int] = mapped_column(default=0)


class Comment(Base):
    __tablename__ = "comments"
    id: Mapped[int] = mapped_column(primary_key=True)
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id"))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    body: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))

Existing endpoints:

• GET /posts — list with pagination
• GET /posts/{id} — detail with comments
• GET /search?q= — basic LIKE search
• GET /posts/popular — top posts
• GET /categories — list

---

Baseline: measure before changing

Critical step: measure performance before any optimization. Without a baseline, you can't show improvements.

Seed realistic data

# scripts/seed.py
import asyncio
from faker import Faker
from datetime import datetime, timezone, timedelta
import random


fake = Faker('es_ES')


async def seed():
    async with SessionLocal() as session:
        # 1000 users
        for i in range(1000):
            session.add(User(email=fake.email(), name=fake.name()))

        # 50 categories (hierarchical)
        # ... generate tree

        # 100,000 posts
        for i in range(100_000):
            session.add(Post(
                author_id=random.randint(1, 1000),
                category_id=random.randint(1, 50),
                title=fake.sentence(),
                body=fake.paragraph(nb_sentences=10),
                published_at=fake.date_time_between(start_date='-2y', end_date='now', tzinfo=timezone.utc),
                view_count=random.randint(0, 10000),
            ))

        # 1,000,000 comments
        # ... many per post
        await session.commit()

Run:

python scripts/seed.py

Measure key queries

# benchmarks/baseline.py
import time
from app.database import SessionLocal
from sqlalchemy import select, text


async def bench_search(q: str = "Python"):
    start = time.perf_counter()
    async with SessionLocal() as session:
        result = await session.execute(
            select(Post).where(
                Post.title.ilike(f"%{q}%") | Post.body.ilike(f"%{q}%")
            ).limit(20)
        )
        result.scalars().all()
    return (time.perf_counter() - start) * 1000


async def bench_top_posts():
    start = time.perf_counter()
    async with SessionLocal() as session:
        # Top posts by view_count
        result = await session.execute(text("""
            SELECT id, title, view_count, COUNT(c.id) AS comment_count
            FROM posts p
            LEFT JOIN comments c ON c.post_id = p.id
            WHERE p.published_at > NOW() - INTERVAL '7 days'
            GROUP BY p.id
            ORDER BY p.view_count DESC
            LIMIT 10
        """))
        result.mappings().all()
    return (time.perf_counter() - start) * 1000


async def bench_post_with_comments(post_id: int = 1):
    start = time.perf_counter()
    async with SessionLocal() as session:
        # Recent comments for post (assume by month)
        result = await session.execute(text("""
            SELECT * FROM comments
            WHERE post_id = :pid
            ORDER BY created_at DESC
            LIMIT 50
        """), {"pid": post_id})
        result.all()
    return (time.perf_counter() - start) * 1000


async def main():
    # Run multiple times, take avg
    print("Search ('Python'):", sum([await bench_search("Python") for _ in range(10)]) / 10, "ms")
    print("Top posts:", sum([await bench_top_posts() for _ in range(10)]) / 10, "ms")
    print("Post with comments:", sum([await bench_post_with_comments() for _ in range(10)]) / 10, "ms")


asyncio.run(main())

Expected output (baseline, before optimizations):

Search ('Python'): 145ms
Top posts: 850ms
Post with comments: 18ms

Write these numbers down. We're going to compare later.

---

Component 1: JSONB metadata on posts (modules 1-2)

Case: posts have variable metadata (SEO tags, social sharing, custom fields). Instead of individual columns, use JSONB.

Migration

# alembic/versions/XXX_add_post_metadata.py
def upgrade() -> None:
    op.add_column(
        'posts',
        sa.Column('metadata', postgresql.JSONB(), nullable=False, server_default='{}')
    )

    # GIN index for queries
    op.execute("""
        CREATE INDEX idx_posts_metadata_gin
        ON posts USING gin (metadata jsonb_path_ops)
    """)


def downgrade() -> None:
    op.drop_index('idx_posts_metadata_gin')
    op.drop_column('posts', 'metadata')

Model

class Post(Base):
    # ... existing fields ...
    metadata: Mapped[dict] = mapped_column(JSONB, default=dict, server_default=text("'{}'"))

Backfill realistic data

# scripts/backfill_metadata.py
async def backfill():
    async with SessionLocal() as session:
        await session.execute(text("""
            UPDATE posts
            SET metadata = jsonb_build_object(
                'seo', jsonb_build_object(
                    'title', title,
                    'description', LEFT(body, 200),
                    'og_image', 'https://example.com/og.jpg'
                ),
                'tags', CASE
                    WHEN id % 3 = 0 THEN '["python", "tutorial"]'::jsonb
                    WHEN id % 3 = 1 THEN '["postgresql", "database"]'::jsonb
                    ELSE '["general"]'::jsonb
                END,
                'reading_time_min', (LENGTH(body) / 1000)::int
            )
        """))
        await session.commit()

Endpoint with metadata queries

@router.get("/posts/by-tag/{tag}")
async def by_tag(tag: str, db = Depends(get_db)):
    """Filter posts by tag using JSONB containment."""
    result = await db.execute(text("""
        SELECT id, title, metadata
        FROM posts
        WHERE metadata->'tags' @> :tag
        ORDER BY published_at DESC
        LIMIT 20
    """), {"tag": f'["{tag}"]'})
    return result.mappings().all()

The idx_posts_metadata_gin index allows a fast query with @>.

Benchmark with JSONB

async def bench_by_tag():
    start = time.perf_counter()
    async with SessionLocal() as session:
        result = await session.execute(text("""
            SELECT id FROM posts
            WHERE metadata->'tags' @> '["python"]'
            LIMIT 20
        """))
        result.all()
    return (time.perf_counter() - start) * 1000

# Without GIN index: ~500ms (seq scan + filter)
# With GIN index: ~5ms

100x improvement. A new capability without adding a column.

---

Component 2: Spanish FTS over title + body (module 3)

Case: replace LIKE search (slow, not language-aware) with full-text search using tsvector.

Migration

# alembic/versions/XXX_add_fts.py
def upgrade() -> None:
    # Generated column with tsvector
    op.execute("""
        ALTER TABLE posts
        ADD COLUMN search_vector tsvector
        GENERATED ALWAYS AS (
            setweight(to_tsvector('spanish', coalesce(title, '')), 'A') ||
            setweight(to_tsvector('spanish', coalesce(body, '')), 'B')
        ) STORED
    """)

    # GIN index for FTS
    op.execute("""
        CREATE INDEX idx_posts_search_vector
        ON posts USING gin (search_vector)
    """)

    # Trigram index for fuzzy fallback
    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
    op.execute("""
        CREATE INDEX idx_posts_title_trgm
        ON posts USING gin (title gin_trgm_ops)
    """)


def downgrade() -> None:
    op.drop_index('idx_posts_title_trgm')
    op.drop_index('idx_posts_search_vector')
    op.execute("ALTER TABLE posts DROP COLUMN search_vector")

Refactored endpoint

@router.get("/search")
async def search_posts(q: str, db = Depends(get_db)):
    """Search with FTS + fuzzy fallback."""
    # FTS
    fts_results = await db.execute(text("""
        SELECT
            id, title,
            ts_rank(search_vector, to_tsquery('spanish', :q)) AS rank
        FROM posts
        WHERE search_vector @@ to_tsquery('spanish', :q)
        ORDER BY rank DESC
        LIMIT 20
    """), {"q": " & ".join(q.split())})

    posts = [dict(r) for r in fts_results.mappings()]

    # If few results, fall back to fuzzy
    suggestions = []
    if len(posts) < 5:
        fuzzy = await db.execute(text("""
            SELECT id, title, similarity(title, :q) AS sim
            FROM posts
            WHERE title % :q
            ORDER BY sim DESC
            LIMIT 5
        """), {"q": q})
        suggestions = [dict(r) for r in fuzzy.mappings()]

    return {"results": posts, "suggestions": suggestions}

FTS benchmark

async def bench_search_fts():
    start = time.perf_counter()
    async with SessionLocal() as session:
        result = await session.execute(text("""
            SELECT id, title FROM posts
            WHERE search_vector @@ to_tsquery('spanish', :q)
            LIMIT 20
        """), {"q": "python"})
        result.all()
    return (time.perf_counter() - start) * 1000

# Before (LIKE): 145ms
# After (FTS + GIN): 8ms

18x faster. Plus better relevance ranking, language-aware (Spanish stems: "programar" matches "programando", "programó", etc.).

---

Component 3: materialized view top_posts_weekly (module 5)

Case: the "popular posts" query with aggregations over comments and views is expensive. Materialize it.

Migration

def upgrade() -> None:
    op.execute("""
        CREATE MATERIALIZED VIEW top_posts_weekly AS
        SELECT
            p.id,
            p.title,
            p.author_id,
            p.published_at,
            p.view_count,
            COUNT(c.id) AS comment_count,
            (p.view_count + COUNT(c.id) * 10) AS popularity_score
        FROM posts p
        LEFT JOIN comments c ON c.post_id = p.id
            AND c.created_at > NOW() - INTERVAL '7 days'
        WHERE p.published_at > NOW() - INTERVAL '7 days'
        GROUP BY p.id
        ORDER BY popularity_score DESC
    """)

    # Unique index for REFRESH CONCURRENTLY
    op.execute("CREATE UNIQUE INDEX ON top_posts_weekly (id)")
    op.execute("CREATE INDEX ON top_posts_weekly (popularity_score DESC)")


def downgrade() -> None:
    op.execute("DROP MATERIALIZED VIEW top_posts_weekly")

Refactored endpoint

@router.get("/posts/popular")
async def popular_posts(db = Depends(get_db)):
    """Top posts this week — uses the MV (instant)."""
    result = await db.execute(text("""
        SELECT id, title, view_count, comment_count, popularity_score
        FROM top_posts_weekly
        ORDER BY popularity_score DESC
        LIMIT 10
    """))
    return result.mappings().all()

Refresh cron

# app/tasks/refresh_top_posts.py
async def refresh():
    async with SessionLocal() as session:
        await session.execute(text("""
            REFRESH MATERIALIZED VIEW CONCURRENTLY top_posts_weekly
        """))
        await session.commit()


# Crontab: every 30min
# */30 * * * * python -m app.tasks.refresh_top_posts

(Capsule 07 will wrap this with an advisory lock.)

Benchmark

# Baseline (direct query with joins): 850ms
# After (MV): 4ms

200x improvement. Trade-off: data up to 30min stale. Acceptable for "popular posts."

---

State at the end of phase 1

So far you have:

• ✅ Component 1: JSONB metadata + GIN index + by_tag endpoint.
• ✅ Component 2: Spanish FTS + pg_trgm fallback + search endpoint.
• ✅ Component 3: materialized view top_posts_weekly + popular endpoint.

Intermediate benchmarks:

Endpoint	Before	After Phase 1
Search (LIKE → FTS)	145ms	8ms (18x)
Popular posts (joined → MV)	850ms	4ms (200x)
By tag (new feature)	N/A	5ms

Suggested commits:

git commit -m "feat: add JSONB metadata to posts with GIN index"
git commit -m "feat: full-text search with tsvector + pg_trgm fallback"
git commit -m "feat: materialized view top_posts_weekly with cron refresh"

---

Pending for phase 2 (capsule 07)

• Component 4: partitioning of comments by month with a zero-downtime migration.
• Component 5: recursive categories with a recursive CTE.
• Component 6: advisory lock on the refresh cron + FTS re-indexing cron.

Plus module 7 extensions if not yet applied (citext on email, etc.).

---

Common traps and mistakes in phase 1

1. Skip the baseline measurement.

Without a baseline, you can't show improvement. Always measure before changing.

2. Migration without a DESCRIPTION.

Document in the commit message what changed and why. It helps future-you and the reviewer.

3. JSONB without an index.

Without a GIN index, JSONB queries are a seq scan. Defeats the purpose of using JSONB.

4. FTS with the wrong language.

to_tsvector('spanish', text)  -- uses Spanish stemming
to_tsvector('english', text)  -- uses English stemming
to_tsvector('simple', text)   -- no stemming, exact words

Match the language of your data.

5. MV without a unique index.

REFRESH MATERIALIZED VIEW CONCURRENTLY requires a unique index. Without it, the refresh is blocking.

6. Final query without an index hint.

Sometimes the planner doesn't use the GIN index. EXPLAIN ANALYZE to verify. If it doesn't, consider adjusting the query or the cost params (capsules 7-8 of #12).

---

Summary and next step

What you have now:

• Project setup and measured baseline.
• Component 1 (JSONB) implemented.
• Component 2 (FTS) implemented.
• Component 3 (MV) implemented.
• Intermediate benchmarks documented.

Before moving on:

• Verify that everything works (tests, manual testing).
• Atomic commits per component.
• Push to the repo.

In the next capsule we complete phase 2: partitioning of comments with a zero-downtime migration, recursive categories with a CTE, and an advisory lock for the crons. Plus the final integration.

---

Resources

• Module 1 (JSONB) and Module 2 (JSONB queries) of this guide.
• Module 3 (FTS) of this guide.
• Module 5 (MVs) of this guide.
• Module 6 (advisory locks) of this guide.

---

Capsule 06 of 08 — Module 8 — Advanced PostgreSQL for Backend Guide