Build 4 — Paginated Search API with Caching

# Build 4 — Paginated Search API ## The brief Build a search API over products: cursor pagination, full-text search (Postgres or Elasticsearch), Redis caching for popular queries. ## Step 1: Setup ```bash uv add fastapi 'uvicorn[standard]' sqlalchemy psycopg2-binary redis ``` ## Step 2: Data model (Postgres with full-text) ```python 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): ```python def upgrade(): op.execute( "CREATE INDEX idx_products_fts ON products USING GIN " "(to_tsvector('english', name || ' ' || description))" ) ``` ## Step 3: Search with cursor pagination ```python 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 ```python 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: ```python @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: ```python 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 ```python 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