| """LibBee FastAPI application for the private Hugging Face origin."""
|
|
|
| import asyncio
|
| import hmac
|
| import logging
|
| from contextlib import asynccontextmanager, suppress |
|
|
| from fastapi import FastAPI, Request
|
| from fastapi.middleware.cors import CORSMiddleware
|
| from fastapi.responses import JSONResponse, Response
|
|
|
| from src.config import LIBBEE_VERSION, get_settings
|
| from src.services.metrics_service import MetricsService
|
| from src.services.rag_service import RAGService
|
|
|
| logging.basicConfig(level=logging.INFO) |
| |
| logging.getLogger("httpx").setLevel(logging.WARNING) |
| logging.getLogger("httpcore").setLevel(logging.WARNING) |
| logger = logging.getLogger(__name__)
|
| settings = get_settings()
|
| rag_service = RAGService()
|
| metrics_service = MetricsService(settings.metrics_path)
|
| _agent_slots = asyncio.Semaphore(settings.max_concurrent_agent_requests) |
| _startup_state = "initializing" |
|
|
|
|
|
|
|
|
| _HF_PREVIEW_PATHS = {"/", "/health/live", "/health/ready"}
|
| _HF_FRAME_ANCESTORS = "frame-ancestors 'self' https://huggingface.co https://*.huggingface.co"
|
|
|
|
|
| def get_rag_service() -> RAGService:
|
| return rag_service
|
|
|
|
|
| def get_metrics_service() -> MetricsService:
|
| return metrics_service
|
|
|
|
|
| def _runtime_config() -> dict:
|
| from src.services.runtime_store import JsonRuntimeStore
|
|
|
| defaults = settings.runtime_defaults |
| try:
|
| return JsonRuntimeStore(settings.config_path, default=defaults).load()
|
| except Exception:
|
| return defaults
|
|
|
|
|
| async def _metrics_flush_loop() -> None:
|
| while True:
|
| await asyncio.sleep(30)
|
| try:
|
| await metrics_service.flush()
|
| except Exception as exc:
|
| logger.warning("Metrics flush error: %s", exc)
|
|
|
|
|
| async def _initialize_services() -> None: |
| global _startup_state |
| from src.services.staff_service import build_staff_index |
|
|
| _startup_state = "initializing" |
| try: |
| await asyncio.to_thread(build_staff_index) |
| except Exception as exc: |
| logger.error("Staff index build failed: %s", exc) |
| try: |
| |
| |
| await rag_service.initialize(openai_api_key=settings.openai_api_key) |
| _startup_state = "ready" if rag_service.is_ready() else "failed" |
| except asyncio.CancelledError: |
| raise |
| except Exception as exc: |
| _startup_state = "failed" |
| logger.error("RAG initialization failed: %s", exc, exc_info=True) |
|
|
|
|
| @asynccontextmanager |
| async def lifespan(_app: FastAPI): |
| |
| |
| startup_task = asyncio.create_task(_initialize_services()) |
| flush_task = asyncio.create_task(_metrics_flush_loop()) |
| try: |
| yield |
| finally: |
| for task in (startup_task, flush_task): |
| task.cancel() |
| for task in (startup_task, flush_task): |
| with suppress(asyncio.CancelledError): |
| await task |
| try: |
| await metrics_service.flush() |
| except Exception as exc: |
| logger.warning("Final metrics flush failed: %s", exc) |
|
|
|
|
| class RequestBodyLimitMiddleware: |
| """Bound received bytes, including chunked bodies without Content-Length.""" |
|
|
| def __init__(self, app, max_bytes: int): |
| self.app = app |
| self.max_bytes = max_bytes |
|
|
| async def __call__(self, scope, receive, send): |
| if scope["type"] != "http" or scope.get("method") not in {"POST", "PUT", "PATCH"}: |
| await self.app(scope, receive, send) |
| return |
|
|
| async def reject(): |
| response = JSONResponse(status_code=413, content={"detail": "Request body too large."}) |
| await response(scope, receive, send) |
|
|
| lengths = [value for name, value in scope.get("headers", []) if name.lower() == b"content-length"] |
| if any(value.isdigit() and int(value) > self.max_bytes for value in lengths): |
| await reject() |
| return |
| chunks = [] |
| total = 0 |
| while True: |
| message = await receive() |
| if message["type"] == "http.disconnect": |
| return |
| chunk = message.get("body", b"") |
| total += len(chunk) |
| if total > self.max_bytes: |
| await reject() |
| return |
| chunks.append(chunk) |
| if not message.get("more_body", False): |
| break |
| body = b"".join(chunks) |
| delivered = False |
|
|
| async def replay(): |
| nonlocal delivered |
| if not delivered: |
| delivered = True |
| return {"type": "http.request", "body": body, "more_body": False} |
| return await receive() |
|
|
| await self.app(scope, replay, send) |
|
|
|
|
| app = FastAPI(
|
| title="LibBee - KU Library AI",
|
| description="Private origin API for the Khalifa University Library AI Assistant",
|
| version=LIBBEE_VERSION,
|
| lifespan=lifespan,
|
| docs_url=None if settings.environment == "production" else "/docs", |
| openapi_url=None if settings.environment == "production" else "/openapi.json", |
| redoc_url=None,
|
| )
|
|
|
| app.add_middleware(
|
| CORSMiddleware,
|
| allow_origins=settings.cors_origins,
|
| allow_credentials=False,
|
| allow_methods=["GET", "POST", "OPTIONS"],
|
| allow_headers=["Content-Type", "X-LibBee-Origin-Token", "X-Request-ID", "X-LibBee-Canary"],
|
| max_age=3600, |
| ) |
| app.add_middleware(RequestBodyLimitMiddleware, max_bytes=settings.max_request_body_bytes) |
|
|
|
|
| @app.middleware("http")
|
| async def secure_origin(request: Request, call_next):
|
| path = request.url.path
|
| protected = any(path == prefix or path.startswith(prefix + "/") for prefix in ("/agent", "/search", "/feedback")) |
| acquired = False
|
|
|
| if path.startswith("/admin") and request.method not in {"GET", "HEAD", "OPTIONS"}:
|
| fetch_site = request.headers.get("sec-fetch-site", "")
|
| if fetch_site and fetch_site not in {"same-origin", "none"}:
|
| return JSONResponse(status_code=403, content={"detail": "Cross-site admin request denied."})
|
|
|
| |
| |
| if protected and request.method != "OPTIONS": |
| supplied = request.headers.get("x-libbee-origin-token", "")
|
| configured = settings.origin_token
|
| local_insecure = settings.environment in {"development", "test"} and not configured
|
| if not local_insecure and (not configured or not hmac.compare_digest(supplied.encode("utf-8"), configured.encode("utf-8"))): |
| return JSONResponse(status_code=401, content={"detail": "Origin authentication required."})
|
|
|
| runtime = _runtime_config()
|
| if runtime.get("maintenance_mode") and request.headers.get("x-libbee-canary") != "1":
|
| return JSONResponse(
|
| status_code=503,
|
| content={"detail": str(runtime.get("maintenance_message") or settings.maintenance_message)[:300]},
|
| )
|
|
|
| if path == "/agent":
|
| try:
|
| await asyncio.wait_for(_agent_slots.acquire(), timeout=0.25)
|
| acquired = True
|
| except TimeoutError:
|
| return JSONResponse(status_code=503, content={"detail": "LibBee is busy. Please retry shortly."})
|
|
|
| try:
|
| response = await call_next(request)
|
| except Exception as exc:
|
| logger.exception("Unhandled application error: %s", exc)
|
| metrics_service.incr_bucket("errors", "unhandled_exception")
|
| response = JSONResponse(status_code=500, content={"detail": "Unexpected server error."})
|
| finally:
|
| if acquired:
|
| _agent_slots.release()
|
|
|
| response.headers["X-Content-Type-Options"] = "nosniff"
|
| response.headers["Referrer-Policy"] = "no-referrer"
|
| response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=(), payment=()"
|
| if path in _HF_PREVIEW_PATHS:
|
|
|
|
|
| response.headers["Content-Security-Policy"] = _HF_FRAME_ANCESTORS
|
| else:
|
| response.headers["X-Frame-Options"] = "SAMEORIGIN" if path.startswith("/admin") else "DENY"
|
| response.headers["Content-Security-Policy"] = (
|
| "frame-ancestors 'self'" if path.startswith("/admin") else "frame-ancestors 'none'"
|
| )
|
| if protected or path.startswith("/admin"):
|
| response.headers["Cache-Control"] = "no-store"
|
| return response
|
|
|
|
|
| from src.api import admin, agent, feedback, search
|
|
|
| app.include_router(agent.router, prefix="/agent", tags=["Agent"])
|
| app.include_router(search.router, prefix="/search", tags=["Search"])
|
| app.include_router(admin.router, prefix="/admin", tags=["Admin"])
|
| app.include_router(feedback.router, prefix="/feedback", tags=["Feedback"])
|
|
|
|
|
| @app.get("/")
|
| @app.get("/health/live")
|
| def health_live():
|
| return {"status": "ok", "version": LIBBEE_VERSION, "service": "LibBee private origin"}
|
|
|
|
|
| @app.get("/health/ready")
|
| def health_ready(): |
| runtime = _runtime_config() |
| rag_ready = rag_service.is_ready() |
| origin_ready = bool(settings.origin_token) or settings.environment in {"development", "test"} |
| if runtime.get("maintenance_mode"): |
| status = "maintenance" |
| elif not origin_ready: |
| status = "misconfigured" |
| elif not rag_ready: |
| status = "initializing" if _startup_state == "initializing" else "unavailable" |
| elif not settings.openai_api_key: |
| status = "degraded" |
| else: |
| status = "ok" |
| return JSONResponse(status_code=200 if status in {"ok", "degraded"} else 503, content={ |
| "status": status, |
| "version": LIBBEE_VERSION, |
| "rag_ready": rag_ready, |
| "retrieval_mode": rag_service.get_stats().get("retrieval_mode", "unavailable"), |
| "model_configured": bool(settings.openai_api_key), |
| "maintenance_mode": bool(runtime.get("maintenance_mode")), |
| }) |
|
|
|
|
| @app.get("/config") |
| def public_config(): |
| runtime = _runtime_config()
|
| return {
|
| "welcome_message": str(runtime.get("welcome_message", ""))[:500],
|
| "announcement": str(runtime.get("announcement", ""))[:500],
|
| "maintenance_mode": bool(runtime.get("maintenance_mode")),
|
| "maintenance_message": str(runtime.get("maintenance_message", ""))[:300],
|
| }
|
|
|
|
|
| @app.get("/favicon.ico", include_in_schema=False)
|
| def favicon():
|
| return Response(status_code=204)
|
|
|
