Build 4 — Paginated Search API with Caching

The brief

Build a search API over products: cursor pagination, full-text search (Postgres or Elasticsearch), Redis caching for popular queries.

Step 1: Setup

uv add fastapi 'uvicorn[standard]' sqlalchemy psycopg2-binary redis

Step 2: Data model (Postgres with full-text)

from sqlalchemy import Index, text
from sqlalchemy.orm import Mapped, mapped_column

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    description: Mapped[str]
    price: Mapped[float]
    category: Mapped[str] = mapped_column(index=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow, index=True)

    __table_args__ = (
        Index("idx_products_fts", text("to_tsvector('english', name || ' ' || description)"), postgresql_using="gin"),
    )

Generated migration (manual; autogenerate misses GIN indexes):

def upgrade():
    op.execute(
        "CREATE INDEX idx_products_fts ON products USING GIN "
        "(to_tsvector('english', name || ' ' || description))"
    )

Step 3: Search with cursor pagination

from pydantic import BaseModel
from typing import Annotated

class ProductResult(BaseModel):
    id: int
    name: str
    price: float
    model_config = ConfigDict(from_attributes=True)

class SearchResponse(BaseModel):
    items: list[ProductResult]
    next_cursor: str | None

@app.get("/search", response_model=SearchResponse)
def search(
    q: str,
    category: str | None = None,
    cursor: str | None = None,
    limit: int = 20,
    db: Session = Depends(get_db),
):
    limit = min(limit, 100)

    stmt = select(Product)
    if q:
        stmt = stmt.where(text("to_tsvector('english', name || ' ' || description) @@ plainto_tsquery(:q)")).params(q=q)
    if category:
        stmt = stmt.where(Product.category == category)
    if cursor:
        cursor_id = int(cursor)
        stmt = stmt.where(Product.id > cursor_id)
    stmt = stmt.order_by(Product.id).limit(limit + 1)

    results = db.execute(stmt).scalars().all()
    has_more = len(results) > limit
    items = results[:limit]
    next_cursor = str(items[-1].id) if has_more else None

    return SearchResponse(items=items, next_cursor=next_cursor)

Cursor pagination is more stable than offset for changing data.

Step 4: Caching

import json
import hashlib
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def cache_key(q, category, cursor, limit):
    raw = f"{q}|{category}|{cursor}|{limit}"
    return f"search:{hashlib.sha256(raw.encode()).hexdigest()}"

@app.get("/search", response_model=SearchResponse)
def search(q: str, category: str | None = None, cursor: str | None = None, limit: int = 20, db: Session = Depends(get_db)):
    key = cache_key(q, category, cursor, limit)
    cached = r.get(key)
    if cached:
        return SearchResponse.model_validate_json(cached)

    # ... do query as before ...
    response = SearchResponse(items=items, next_cursor=next_cursor)
    r.setex(key, 300, response.model_dump_json())    # 5 min TTL
    return response

Cache popular queries; invalidate via TTL.

Step 5: Cache invalidation

When products change, invalidate related caches:

@app.post("/products")
def create_product(payload: ProductCreate, ...):
    product = Product(**payload.model_dump())
    db.add(product); db.commit()

    # Invalidate: simplest is to clear all search caches
    for key in r.scan_iter("search:*"):
        r.delete(key)

    return product

For more granular: scope invalidation (only the category's caches).

Step 6: Stampede protection

If many requests hit a popular search at once with a cache miss:

def search_with_lock(key, compute_func):
    cached = r.get(key)
    if cached:
        return cached

    lock_key = f"lock:{key}"
    if r.set(lock_key, "1", nx=True, ex=10):
        try:
            result = compute_func()
            r.setex(key, 300, result)
            return result
        finally:
            r.delete(lock_key)
    else:
        # Wait briefly for the in-flight compute
        time.sleep(0.1)
        cached = r.get(key)
        return cached or compute_func()

Step 7: Tests

def test_search_basic(client, db):
    db.add(Product(name="iPhone 15", description="Apple phone", price=999, category="electronics"))
    db.commit()
    response = client.get("/search?q=iPhone")
    assert response.status_code == 200
    assert len(response.json()["items"]) == 1

def test_pagination(client, db):
    for i in range(30):
        db.add(Product(name=f"Product {i}", description="", price=i, category="test"))
    db.commit()
    r1 = client.get("/search?q=Product&limit=10")
    assert len(r1.json()["items"]) == 10
    assert r1.json()["next_cursor"] is not None

    cursor = r1.json()["next_cursor"]
    r2 = client.get(f"/search?q=Product&limit=10&cursor={cursor}")
    assert len(r2.json()["items"]) == 10

Deliverables

• Search endpoint with Postgres full-text.
• Cursor pagination.
• Redis caching with TTL.
• Cache invalidation on writes.
• Stampede protection.
• Tests covering search + pagination + caching.

What you'll learn

• Cursor vs offset pagination (when each fits).
• Postgres full-text search (basic).
• Redis as a cache; key design.
• Cache invalidation strategies.
• Stampede protection patterns.

Stretch

• Switch to Elasticsearch for richer search (typo tolerance, faceted).
• Search analytics (most-popular queries).
• Multi-tenant: tenant_id in cache key.
• Personalized ranking (boost recent views).

---

📊 100-Point Evaluation Rubric (Case 4)

Evaluation Dimension	Weight	Target Standard (Pass ≥ 80%)
1. Search Correctness	40 pts	Full-text search over name+description (GIN index via manual migration — autogenerate misses it); category filter; correct results ordering.
2. Pagination + Cache	30 pts	Cursor (not offset) pagination with next_cursor; Redis cache keyed on (q, category, cursor, limit) with invalidation story.
3. Docs + OpenAPI	15 pts	/docs schemas for search request/response; README explains cursor vs offset choice.
4. Code Quality	15 pts	limit clamped (≤100), empty-page edge handled, no N+1 on result shaping.

---

📤 Submission Instructions & Guidelines

1. Deliverable Formats:
   • Option A (Recommended): Upload a single .zip archive containing your project directory: src/, tests/, Dockerfile (where applicable), and README.md with run instructions.
   • Option B: Upload your README.md and include your GitHub repository link in the Submission Notes box below.
2. Pre-Submission Verification:
   • Run your full test suite (pytest) green; include the passing output in your README.
   • Exercise every endpoint once via /docs (Swagger UI) and paste one sample request/response per endpoint into your README.
   • Confirm no secrets are committed (.env git-ignored; secrets via environment only).