# services/api/main.py import asyncio import json import logging import os import time from contextlib import asynccontextmanager import asyncpg import httpx import redis.asyncio as aioredis from fastapi import FastAPI, Request, Response from kafka import KafkaProducer as SyncKafkaProducer from prometheus_client import ( CONTENT_TYPE_LATEST, REGISTRY, Counter, Gauge, Histogram, generate_latest, ) from routers import events, health, movies, recommend, simulate, ws import tmdb logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379") POSTGRES_URL = os.getenv("POSTGRES_URL", "postgresql://recsys:recsys@postgres:5432/recsys") KAFKA_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:9092") def _get_or_create(cls, name: str, doc: str, *args, fallback_suffix: str = "", **kwargs): try: return cls(name, doc, *args, **kwargs) except ValueError: key = f"{name}{fallback_suffix}" if fallback_suffix else name return REGISTRY._names_to_collectors.get(key) or REGISTRY._names_to_collectors.get(name) REQUEST_LATENCY = _get_or_create( Histogram, "api_latency_ms", "API request latency in milliseconds", ["method", "endpoint"], fallback_suffix="_bucket", buckets=[1, 5, 10, 25, 50, 100, 250, 500, 1000], ) CACHE_HIT_COUNTER = _get_or_create( Counter, "cache_hit_total", "Redis cache hits", ["result"], fallback_suffix="_total", ) DLQ_GAUGE = _get_or_create( Gauge, "recsys_dlq_events_total", "Total invalid events in dead-letter queue", ) MODEL_NDCG_GAUGE = _get_or_create( Gauge, "recsys_model_ndcg_at_10", "ALS model NDCG@10 from last bootstrap run", ) async def _prewarm_tmdb(app: FastAPI) -> None: """Background task: warm the Redis TMDB cache for the most popular movies so posters & descriptions render instantly on first browse/recommend. Idempotent (enrich skips already-cached entries), so it's cheap on every restart.""" log = logging.getLogger("tmdb.prewarm") try: await asyncio.sleep(3) # let startup settle pg, redis, http = app.state.pg_pool, app.state.redis, app.state.http if pg is None: return async with pg.acquire() as conn: rows = await conn.fetch( "SELECT movie_id, title, genres, year FROM movies " "ORDER BY popularity DESC NULLS LAST LIMIT 600" ) items = [dict(r) for r in rows] for i in range(0, len(items), 10): # 10 concurrent to stay under TMDB rate limit await tmdb.enrich(redis, http, items[i:i + 10]) await asyncio.sleep(0.25) # ~40 req/s TMDB free-tier limit log.info("Warmed TMDB cache for %d popular movies", len(items)) except asyncio.CancelledError: pass except Exception as e: # never let prewarm crash the app log.warning("TMDB prewarm skipped: %s", e) @asynccontextmanager async def lifespan(app: FastAPI): app.state.redis = aioredis.from_url(REDIS_URL, decode_responses=False) app.state.pg_pool = await asyncpg.create_pool(POSTGRES_URL, min_size=2, max_size=10) loop = asyncio.get_running_loop() app.state.kafka_producer = await loop.run_in_executor( None, lambda: SyncKafkaProducer( bootstrap_servers=KAFKA_SERVERS.split(","), value_serializer=lambda v: json.dumps(v).encode(), key_serializer=lambda k: str(k).encode(), ) ) app.state.http = httpx.AsyncClient(timeout=10) prewarm = asyncio.create_task(_prewarm_tmdb(app)) yield prewarm.cancel() await app.state.http.aclose() await app.state.redis.aclose() await app.state.pg_pool.close() app.state.kafka_producer.close() def create_app(lifespan_fn=None) -> FastAPI: """Create FastAPI app. Called both in production (with lifespan) and tests (state injected).""" app = FastAPI(title="Movie RecSys API", lifespan=lifespan_fn) app.include_router(recommend.router) app.include_router(simulate.router) app.include_router(events.router) app.include_router(health.router) app.include_router(movies.router) app.include_router(ws.router) @app.middleware("http") async def metrics_middleware(request: Request, call_next): t0 = time.monotonic() response = await call_next(request) latency_ms = (time.monotonic() - t0) * 1000 if REQUEST_LATENCY is not None: REQUEST_LATENCY.labels( method=request.method, endpoint=request.url.path, ).observe(latency_ms) return response @app.get("/metrics") async def metrics(request: Request): # These gauges are owned by Spark/bootstrap, not the API — pull from Redis on each scrape. r = request.app.state.redis dlq_raw, ml_raw = await r.mget("dlq:count", "eval:als_metrics") if DLQ_GAUGE is not None: DLQ_GAUGE.set(int(dlq_raw or 0)) if ml_raw and MODEL_NDCG_GAUGE is not None: MODEL_NDCG_GAUGE.set(json.loads(ml_raw).get("ndcg@10", 0.0)) return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST) @app.get("/stats") async def get_stats(request: Request): r = request.app.state.redis vals = await r.mget( "stats:cache_hits", "stats:requests_total", "stats:events_total", "stats:last_latency_ms", "dlq:count", ) hits, total, evts = int(vals[0] or 0), int(vals[1] or 0), int(vals[2] or 0) latency, dlq = float(vals[3] or 0), int(vals[4] or 0) return { "cache_hit_rate": round(hits / total, 2) if total > 0 else 0.0, "p50_latency_ms": latency, "total_events": evts, "dlq_count": dlq, } @app.get("/events/dlq") async def dlq_events(request: Request): """Return up to 10 most recent invalid events routed to the dead-letter queue.""" items = await request.app.state.redis.lrange("dlq:recent", 0, 9) return [json.loads(i) for i in items] @app.get("/stats/ml") async def ml_stats(request: Request): """Return offline ALS evaluation metrics (written by bootstrap job).""" raw = await request.app.state.redis.get("eval:als_metrics") if not raw: return {"error": "No evaluation data — run `make bootstrap` to generate metrics."} return json.loads(raw) @app.get("/stats/ab") async def ab_stats(request: Request): """A/B experiment: control = ALS baseline order, treatment = personalized hybrid.""" r = request.app.state.redis vals = await r.mget( "ab:impressions:control", "ab:impressions:treatment", "ab:engagements:control", "ab:engagements:treatment", ) a_imp, b_imp, a_eng, b_eng = [int(v or 0) for v in vals] return { "config": { "control": "ALS baseline order (no event-driven personalization)", "treatment": "Personalized (Spark time-decay + hybrid CF+genre)", }, "control": { "impressions": a_imp, "engagements": a_eng, "engagement_rate": round(a_eng / a_imp, 4) if a_imp else 0.0, }, "treatment": { "impressions": b_imp, "engagements": b_eng, "engagement_rate": round(b_eng / b_imp, 4) if b_imp else 0.0, }, } @app.get("/stats/model-history") async def model_history(request: Request): """Return last 10 ALS training runs with eval metrics from model_runs table.""" rows = await request.app.state.pg_pool.fetch( "SELECT id, trained_at, rank, max_iter, reg_param, seed, " "ndcg_10, precision_10, n_users_eval, is_active " "FROM model_runs ORDER BY trained_at DESC LIMIT 10" ) return [dict(r) for r in rows] return app # Production entry: app with lifespan app = create_app(lifespan_fn=lifespan) if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=int(os.getenv("API_PORT", "8000")), reload=False)