Module 5: Materialized Views

Module 5 deliverable: blog dashboard with materialized views

What are you going to build and why?

You're going to implement the complete system of a blog's internal dashboard using 4 materialized views. The dashboard serves analytical metrics (weekly top posts, monthly top commenters, posts-per-month trend, most active categories) that without precomputation take ~17 seconds in total to load. With well-designed MVs, it drops to ~50 milliseconds. The refresh is scheduled via cron, the frontend queries are instant lookups over indexed tables, and the "last updated" banner communicates the staleness to the user.

This project integrates everything you learned in the module: the decision of when to use an MV (lessons 02 and 07), creation with WITH NO DATA for fast deploys (lesson 03), refresh CONCURRENTLY with the mandatory unique index (lesson 04), appropriate indexes according to the queries (lesson 05), and the "one MV per panel" architecture (lesson 06).

Why this project is portfolio-worthy: it delivers a realistic PR with Alembic migrations, FastAPI/SQLAlchemy 2.0 async code, tests, a cron job, and architectural documentation (ADR). A senior technical reviewer can assess your level by looking at: (a) whether you chose MV correctly for each panel; (b) whether the unique indexes are there from day 1; (c) whether the query indexes cover the real patterns; (d) whether the staleness banner is in the endpoints; (e) whether the cron is protected (which we anticipate for module 6).

Connection to the capstone project (module 8): what you deliver here is functionally identical to the "Materialized view top_posts_weekly" component of the final project. Module 6 adds pg_try_advisory_lock for safe refresh against concurrency. Module 8 integrates it into the refactor of the complete Blog API with FTS, partitioning, JSONB, and recursive CTEs.

Project objective

By completing this project, you'll have:

4 materialized views created with Alembic migrations (1 per dashboard panel).
A cron job that refreshes the MVs with appropriate frequencies.
4 FastAPI endpoints that consume the MVs and return last_updated + staleness_minutes.
1 aggregator endpoint GET /dashboard that returns the 4 panels together.
Tests for the endpoints (minimum: correct schema, staleness present).
A documented benchmark: before/after latency of each panel.
An ARCHITECTURE.md with justified decisions (why MV and not Redis for each panel).

How it fits with what you learned

Module concept	Where it's used in the project
Lesson 02: views vs materialized views	Decision of MV vs direct query vs Redis for each panel — documented in `ARCHITECTURE.md`.
Lesson 03: creation and refresh	Each migration creates an MV with `CREATE MATERIALIZED VIEW ... WITH NO DATA` + the mandatory unique index + first `FULL` refresh.
Lesson 04: `CONCURRENTLY` vs `FULL`	The cron always uses `CONCURRENTLY`. The migration uses `FULL` for the first refresh post-`WITH NO DATA`.
Lesson 05: indexes on MVs	Each MV has appropriate indexes for its endpoint's queries.
Lesson 06: analytics use cases	"One MV per panel" pattern + aggregator endpoint + `last_updated` banner.
Lesson 07: decision matrix	The `ARCHITECTURE.md` applies the matrix to justify each decision.

Think of the project as the minimum viable system of an internal analytics dashboard. Each piece has a purpose; nothing is filler.

Technical specifications

Stack

Language: Python 3.11+
Main framework: FastAPI 0.110+
ORM: SQLAlchemy 2.0+ async
DB: PostgreSQL 16+ (assuming posts, views, comments, users, categories already exist)
Migrations: Alembic
Tests: pytest + httpx (for async tests)
Cron: OS cron (*/60 * * * *) or pg_cron (optional)

Initial setup

Assuming you already have the Blog API project (from guide #8) with tables posts, views, comments, users, categories. If not, you can use the setup from lesson 03 to have minimal tables.

# Project structure (relevant excerpt)
app/
├── api/
│   └── dashboard.py          # endpoints (this project)
├── jobs/
│   └── refresh_dashboard.py  # cron job (this project)
├── models/
│   └── mv.py                  # mappings of the 4 MVs (this project)
├── schemas/
│   └── dashboard.py           # Pydantic schemas (this project)
├── services/
│   └── mv_refresh.py          # refresh function with whitelist (this project)
└── alembic/
    └── versions/
        ├── 042_mv_top_posts_weekly.py
        ├── 043_mv_top_commenters_monthly.py
        ├── 044_mv_posts_per_month.py
        └── 045_mv_active_categories_weekly.py

ARCHITECTURE.md               # decisions document (this project)

Mandatory functionalities

1. MV `mv_top_posts_weekly`

Spec:

10 rows (top posts by views last 7 days).
Columns: post_id, title, slug, view_count, computed_at.
Unique index on post_id.
Refresh frequency: every 1 hour.

Expected migration (042_mv_top_posts_weekly.py):

"""create mv_top_posts_weekly

Revision ID: 042_mv_top_posts_weekly
"""
from alembic import op


def upgrade() -> None:
    op.execute("""
        CREATE MATERIALIZED VIEW mv_top_posts_weekly AS
        SELECT
            p.id AS post_id,
            p.title,
            p.slug,
            count(v.id) AS view_count,
            NOW() AS computed_at
        FROM posts p
        JOIN views v ON v.post_id = p.id
        WHERE v.created_at > NOW() - INTERVAL '7 days'
          AND p.published_at IS NOT NULL
        GROUP BY p.id, p.title, p.slug
        ORDER BY view_count DESC
        LIMIT 10
        WITH NO DATA;
    """)
    op.execute("""
        CREATE UNIQUE INDEX idx_mv_top_posts_weekly_pk
            ON mv_top_posts_weekly (post_id);
    """)
    # First refresh (FULL — necessary after WITH NO DATA)
    op.execute("REFRESH MATERIALIZED VIEW mv_top_posts_weekly;")


def downgrade() -> None:
    op.execute("DROP MATERIALIZED VIEW IF EXISTS mv_top_posts_weekly;")

Endpoint: GET /dashboard/top-posts-weekly

Expected response:

{
  "items": [
    {"post_id": 142, "title": "Post X", "slug": "post-x", "view_count": 8421},
    ...
  ],
  "last_updated": "2026-05-02T14:30:00Z",
  "staleness_minutes": 47
}

2. MV `mv_top_commenters_monthly`

Spec:

10 rows (top commenters by # comments last 30 days).
Columns: user_id, username, comment_count, computed_at.
Unique index on user_id.
Refresh frequency: every 1 hour.

Endpoint: GET /dashboard/top-commenters-monthly

3. MV `mv_posts_per_month`

Spec:

12 rows (one row per month in the last 12 months).
Columns: month, post_count, computed_at.
Unique index on month.
Refresh frequency: every 6 hours (changes little hour to hour).

Endpoint: GET /dashboard/posts-per-month

4. MV `mv_active_categories_weekly`

Spec:

N rows (all categories ranked, top 10 shown).
Columns: category_id, category_name, view_count, comment_count, activity_score, computed_at.
Unique index on category_id.
Additional index on activity_score DESC for the endpoint's ORDER BY.
Refresh frequency: every 1 hour.

Endpoint: GET /dashboard/active-categories-weekly

5. Aggregator endpoint

Spec: GET /dashboard returns the 4 panels in a single request, with overall_last_updated (the oldest of the 4) and overall_staleness_minutes (the worst of the 4).

{
  "top_posts_weekly": { ... },
  "top_commenters_monthly": { ... },
  "posts_per_month": { ... },
  "active_categories_weekly": { ... },
  "overall_last_updated": "2026-05-02T14:00:00Z",
  "overall_staleness_minutes": 77
}

6. Refresh cron

Spec: a Python script that refreshes the 4 MVs with CONCURRENTLY, records duration and status, and runs via OS cron or pg_cron.

# jobs/refresh_dashboard.py
import asyncio
import logging
from app.db import async_session_factory
from services.mv_refresh import refresh_mv

logger = logging.getLogger(__name__)


MVS_HOURLY = [
    "mv_top_posts_weekly",
    "mv_top_commenters_monthly",
    "mv_active_categories_weekly",
]


async def refresh_hourly_mvs() -> None:
    """Refresh the MVs that update every hour."""
    async with async_session_factory() as session:
        for mv in MVS_HOURLY:
            result = await refresh_mv(session, mv, concurrent=True)
            if result["success"]:
                logger.info(f"Refreshed {mv} in {result['duration_ms']}ms")
            else:
                logger.error(f"Failed {mv}: {result.get('error')}")


async def refresh_six_hourly_mvs() -> None:
    """Refresh the MVs that update every 6 hours."""
    async with async_session_factory() as session:
        result = await refresh_mv(session, "mv_posts_per_month", concurrent=True)
        if result["success"]:
            logger.info(f"Refreshed mv_posts_per_month in {result['duration_ms']}ms")
        else:
            logger.error(f"Failed: {result.get('error')}")


if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "six-hourly":
        asyncio.run(refresh_six_hourly_mvs())
    else:
        asyncio.run(refresh_hourly_mvs())

OS cron:

# /etc/cron.d/dashboard-refresh
0 * * * * cd /app && python -m jobs.refresh_dashboard >> /var/log/refresh.log 2>&1
0 */6 * * * cd /app && python -m jobs.refresh_dashboard six-hourly >> /var/log/refresh.log 2>&1

Validations and error handling

What must be validated

Each MV has a unique index — if it's missing, REFRESH CONCURRENTLY fails.
The endpoint handles an empty MV (returns an empty list with staleness_minutes = 0).
The endpoint uses timezone-aware datetimes (not naive).
Whitelist of MVs in refresh_mv (prevents SQL injection).

Errors that must be handled

MV not populated (first refresh failed): the endpoint must respond with something reasonable (empty list + error log), not crash.
Refresh fails due to contention: the cron logs the error without aborting the following MVs.
Missing unique index: detect it in the refresh_mv code and return an actionable message ("MV needs UNIQUE INDEX for concurrent refresh").

Minimal implementation example

A functional runnable skeleton. It's up to you to expand it to meet the mandatory functionalities.

# app/api/dashboard.py — minimal skeleton
from datetime import datetime, timezone
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.deps import get_session
from app.models.mv import TopPostsWeekly  # mapped to mv_top_posts_weekly

router = APIRouter(prefix="/dashboard", tags=["dashboard"])


class TopPostItem(BaseModel):
    post_id: int
    title: str
    slug: str
    view_count: int


class TopPostsResponse(BaseModel):
    items: list[TopPostItem]
    last_updated: datetime
    staleness_minutes: int


@router.get("/top-posts-weekly", response_model=TopPostsResponse)
async def get_top_posts_weekly(
    session: AsyncSession = Depends(get_session),
) -> TopPostsResponse:
    stmt = select(TopPostsWeekly).order_by(TopPostsWeekly.view_count.desc())
    result = await session.execute(stmt)
    rows = result.scalars().all()

    if not rows:
        now = datetime.now(timezone.utc)
        return TopPostsResponse(items=[], last_updated=now, staleness_minutes=0)

    last_updated = rows[0].computed_at
    now = datetime.now(timezone.utc)
    staleness = int((now - last_updated).total_seconds() / 60)

    return TopPostsResponse(
        items=[
            TopPostItem(
                post_id=r.post_id, title=r.title, slug=r.slug, view_count=r.view_count
            )
            for r in rows
        ],
        last_updated=last_updated,
        staleness_minutes=max(staleness, 0),
    )

# TODO: implement endpoints for the other 3 panels
# TODO: implement the aggregator endpoint GET /dashboard

# app/services/mv_refresh.py — skeleton
import time
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession


ALLOWED_MVS = {
    "mv_top_posts_weekly",
    "mv_top_commenters_monthly",
    "mv_posts_per_month",
    "mv_active_categories_weekly",
}


async def refresh_mv(
    session: AsyncSession,
    mv_name: str,
    *,
    concurrent: bool = True,
) -> dict:
    if mv_name not in ALLOWED_MVS:
        return {"success": False, "error": f"MV '{mv_name}' not allowed"}

    mode = "CONCURRENTLY " if concurrent else ""
    sql = f"REFRESH MATERIALIZED VIEW {mode}{mv_name}"

    started = time.monotonic()
    try:
        await session.execute(text(sql))
        await session.commit()
        duration_ms = (time.monotonic() - started) * 1000
        return {"success": True, "duration_ms": round(duration_ms, 2)}
    except SQLAlchemyError as e:
        await session.rollback()
        return {"success": False, "error": str(e)}

# TODO: detect the specific "needs unique index" error
# TODO: implement fallback to FULL if the MV is not populated (lesson 04)

This skeleton:

Is runnable (you can run the endpoint today).
Shows the expected structure (timezone-aware, schema with metadata, whitelist).
Does NOT include: the other 3 endpoints, the cron, the aggregator endpoint, the tests, the documentation. That's your work.

Evaluation rubric (self-check)

Total: 100 points. Passing: ≥70 points.

Functionality (50 points)

(10 pts) MV mv_top_posts_weekly created with migration + unique index + first refresh.
(10 pts) MV mv_top_commenters_monthly created with migration + unique index + first refresh.
(5 pts) MV mv_posts_per_month created with migration + unique index + first refresh.
(10 pts) MV mv_active_categories_weekly created with migration + unique index + additional index on activity_score DESC.
(5 pts) 4 individual endpoints return the correct schema + last_updated + staleness_minutes.
(5 pts) Aggregator endpoint GET /dashboard returns the 4 panels + overall_*.
(5 pts) Cron job runs the refresh of the 4 MVs without errors under normal conditions.

Code quality (25 points)

(5 pts) Consistent type hints in Python code (Pydantic + Mapped in SQLAlchemy).
(5 pts) Whitelist of MVs in refresh_mv (doesn't accept an arbitrary name).
(5 pts) Reasonable error handling (try/except + rollback + structured log).
(5 pts) Timezone-aware datetimes throughout the code.
(5 pts) Tests of the endpoints (minimum: status 200 + presence of required fields).

Documentation (15 points)

(5 pts) ARCHITECTURE.md justifies each MV with quantitative criteria (target latency, acceptable staleness, query frequency).
(5 pts) Documented benchmark: before/after latency of each panel with EXPLAIN ANALYZE before and EXPLAIN ANALYZE after.
(5 pts) README explains how to run the cron (OS cron or pg_cron).

Extra credit (optional, up to +10 pts)

(+3 pts) Endpoint that exposes refresh metrics (duration, last success, last failure) at /admin/mv-status.
(+3 pts) Integration test that verifies: refresh → change in base data → refresh again → endpoint returns updated data.
(+2 pts) Implementation of fallback to FULL when the MV is not populated (lesson 04).
(+2 pts) Mock banner on the frontend (simple HTML/JS) that consumes the endpoint and shows "Last updated: X minutes ago".

Common mistakes in this project

Mistake 1: forgetting the unique index in some migration

Symptom: the migration runs OK, the first FULL refresh too, but the cron with CONCURRENTLY fails with cannot refresh materialized view ... concurrently.

Why it happens: you forgot CREATE UNIQUE INDEX in the migration. It's the #1 error in this project.

How to fix:

# in the migration, IMMEDIATELY after the CREATE MATERIALIZED VIEW:
op.execute("""
    CREATE UNIQUE INDEX idx_<mv_name>_pk
        ON <mv_name> (<unique_column>);
""")

Audit your 4 migrations before merging.

Mistake 2: the endpoint fails with an empty MV (first deploy)

Symptom: after a migration with WITH NO DATA, the endpoint returns 500 because rows[0].computed_at fails with "list index out of range."

Why it happens: the MV is empty (it hasn't refreshed yet), but the endpoint assumes there's always at least 1 row.

How to fix: check if not rows before accessing rows[0]. The skeleton I gave you already does it correctly, but make sure to keep it when expanding.

Mistake 3: naive vs timezone-aware datetime

Symptom: error can't compare offset-naive and offset-aware datetimes when you compute staleness_minutes.

Why it happens: datetime.utcnow() returns naive, computed_at from the DB returns aware (because the column is TIMESTAMPTZ). Subtracting them fails.

How to fix: always use datetime.now(timezone.utc):

from datetime import datetime, timezone

now = datetime.now(timezone.utc)
staleness = (now - last_updated).total_seconds() / 60

Never use datetime.utcnow() (deprecated in Python 3.12 and prone to this bug).

Mistake 4: cron without a whitelist of MVs

Symptom: the cron accepts an arbitrary MV name via argument. A typo runs REFRESH MATERIALIZED VIEW some_typo which fails with an SQL syntax error.

Why it happens: without a whitelist, the code is flexible but permissive. More serious: if the name comes from uncontrolled input, it's an SQL injection vulnerability (table names can't be parameterized).

How to fix: an explicit whitelist in ALLOWED_MVS. Only names on that list are accepted.

Mistake 5: sequential refresh when it could be parallel

Symptom: the cron takes 17 seconds (the sum of the 4 refreshes). In parallel it would take ~9s (the slowest, mv_active_categories_weekly).

Why it happens: a naive implementation loops the MVs sequentially.

How to fix: use asyncio.gather with independent sessions:

async def refresh_in_parallel() -> list[dict]:
    async def refresh_one(mv_name):
        async with async_session_factory() as session:
            return await refresh_mv(session, mv_name, concurrent=True)

    results = await asyncio.gather(*[refresh_one(mv) for mv in MVS_HOURLY])
    return results

Trade-off: more simultaneous connections to Postgres. If your pool has a low cap, consider it.

Mistake 6: forgetting `commit()` after the refresh

Symptom: the refresh "seems" to work but the data doesn't change from the session that ran the refresh.

Why it happens: SQLAlchemy 2.0 async operates in transactions. Without commit(), other sessions don't see the change.

How to fix: always await session.commit() after the refresh. It's in the skeleton I gave you — keep it.

Mistake 7: `overall_last_updated` computed backwards

Symptom: the dashboard shows overall_staleness_minutes: 0 even though one of the panels has 60 minutes of staleness.

Why it happens: someone used max(last_updated) (the most recent) instead of min(last_updated) (the oldest).

How to fix: logically — the "overall last updated" is the moment from which some data may be out of date, which is the oldest of the timestamps:

overall_last_updated = min(all_last_updated)
overall_staleness_minutes = max(all_staleness)

What to do if you get stuck

Table setup doesn't work: review lesson 03, the "Setup of tables and test data" section. Make sure posts, views, comments, users, categories exist.
CREATE MATERIALIZED VIEW fails: review the exact syntax in lesson 03. The usual error is a nonexistent column or a misspelled alias.
REFRESH CONCURRENTLY fails: lesson 04. It's almost always the unique index.
The endpoint works but returns old data after the refresh: forgot the commit() or you're querying an old session. Review lesson 03.
The cron doesn't run: if you use OS cron, validate with crontab -l. To debug, run the script manually: python -m jobs.refresh_dashboard.
Tests fail in CI but pass locally: make sure the test database has the MVs created (run migrations before the tests).

Resources for the project

PostgreSQL 16 — Materialized Views — complete official reference.
SQLAlchemy 2.0 — async patterns — for async sessions and queries.
Alembic — autogenerate vs manual migrations — to understand why MVs require op.execute() (Alembic doesn't autodetect MVs).
Crunchy Data — Materialized Views Best Practices — operational checklist.
pganalyze — Monitoring Materialized View Refreshes — for the metrics extra credit.
pg_cron repo — alternative option to OS cron.
FastAPI — testing async endpoints — for the tests with httpx.

What's next

What you built here is the direct base of the "Materialized view top_posts_weekly" component of the final project of module 8. You carry that piece over to the refactor of the complete Blog API almost unchanged.

But before that there's a critical intermediate module. Your current cron doesn't protect against concurrent refreshes — if two instances start at the same time (double deploy, k8s reconciling), the second hangs waiting for the first. Module 6 teaches you pg_try_advisory_lock — a non-transactional lock that lets the second cron skip instead of waiting. It's the last piece to make your system production-grade. The narrative transition: "you have a cron that refreshes every hour. But what happens if two instances start at the same time? For that, PostgreSQL has advisory locks — non-transactional locks that replace Redis SETNX for distributed coordination."

Before moving on to module 6, make sure your project:

Passes all the tests.
Meets ≥70 points on the rubric.
Has an ARCHITECTURE.md justifying each MV with the decision matrix from lesson 07.
Has a documented before/after benchmark.
Runs the cron successfully at least 2 times in a row (to validate idempotency).

When it's ready, open module 6.

Module 5 — Advanced PostgreSQL for Backend Guide

Next module: Advisory Locks + Savepoints — distributed coordination without Redis and partial rollbacks with savepoints.

← PreviousMVs vs application cache (Redis): decision matrix with quantitative criteria Next →Module 6: Advisory Locks + Savepoints

Module 5 deliverable: blog dashboard with materialized views

What are you going to build and why?

Project objective

How it fits with what you learned

Technical specifications

Stack

Initial setup

Mandatory functionalities

1. MV mv_top_posts_weekly

2. MV mv_top_commenters_monthly

3. MV mv_posts_per_month

4. MV mv_active_categories_weekly

5. Aggregator endpoint

6. Refresh cron

Validations and error handling

What must be validated

Errors that must be handled

Minimal implementation example

Evaluation rubric (self-check)

Functionality (50 points)

Code quality (25 points)

Documentation (15 points)

Extra credit (optional, up to +10 pts)

Common mistakes in this project

Mistake 1: forgetting the unique index in some migration

Mistake 2: the endpoint fails with an empty MV (first deploy)

Mistake 3: naive vs timezone-aware datetime

Mistake 4: cron without a whitelist of MVs

Mistake 5: sequential refresh when it could be parallel

Mistake 6: forgetting commit() after the refresh

Mistake 7: overall_last_updated computed backwards

What to do if you get stuck

Resources for the project

What's next

1. MV `mv_top_posts_weekly`

2. MV `mv_top_commenters_monthly`

3. MV `mv_posts_per_month`

4. MV `mv_active_categories_weekly`

Mistake 6: forgetting `commit()` after the refresh

Mistake 7: `overall_last_updated` computed backwards