Create app.py

app.py

@@ -0,0 +1,86 @@
+from fastapi import FastAPI, HTTPException, Header, Depends, Request
+from pydantic import BaseModel
+from transformers import pipeline
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+import logging
+import os
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+limiter = Limiter(key_func=get_remote_address)
+
+app = FastAPI(title="Panoptifi Topics API")
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+API_KEY = os.environ.get("API_KEY", "")
+
+
+def verify_api_key(x_api_key: str = Header(None, alias="X-API-Key")):
+  if API_KEY and x_api_key != API_KEY:
+      raise HTTPException(status_code=401, detail="Invalid API key")
+  return True
+
+
+logger.info("Loading zero-shot classifier...")
+classifier = pipeline(
+  "zero-shot-classification",
+  model="MoritzLaurer/ModernBERT-large-zeroshot-v2.0"
+)
+logger.info("Model loaded")
+
+DEFAULT_LABELS = [
+  "monetary policy",
+  "earnings",
+  "mergers and acquisitions",
+  "regulation",
+  "layoffs",
+  "product launch",
+  "legal issues",
+  "market sentiment",
+  "cryptocurrency",
+  "economic data"
+]
+
+
+class TopicInput(BaseModel):
+  text: str
+  labels: list[str] | None = None
+  multi_label: bool = False
+
+
+class TopicScore(BaseModel):
+  label: str
+  score: float
+
+
+class TopicResult(BaseModel):
+  labels: list[TopicScore]
+
+
+@app.get("/health")
+@limiter.limit("60/minute")
+def health(request: Request):
+  return {"status": "healthy", "model": "ModernBERT-large-zeroshot-v2.0"}
+
+
+@app.post("/classify", response_model=TopicResult)
+@limiter.limit("30/minute")
+def classify_topic(request: Request, input: TopicInput, _: bool = Depends(verify_api_key)):
+  if not input.text.strip():
+      raise HTTPException(400, "Text cannot be empty")
+
+  labels = input.labels or DEFAULT_LABELS
+  result = classifier(
+      input.text[:2000],
+      labels,
+      multi_label=input.multi_label
+  )
+
+  return TopicResult(labels=[
+      TopicScore(label=label, score=score)
+      for label, score in zip(result["labels"], result["scores"])
+  ])