"""FastAPI application factory. Long-lived services (manifest, indexes, providers) are created in the lifespan handler and attached to ``app.state`` so every route shares one consistent view of the corpus. Importing this module performs no I/O. """ from __future__ import annotations import logging import time import uuid from contextlib import asynccontextmanager from pathlib import Path from typing import Optional from fastapi import FastAPI, HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse, Response from ..core.config import Settings, get_settings from ..core.types import IndexCompatibilityError, IngestionError, NotFoundError from ..monitoring import metrics from ..rag.generator import Generator, ProviderError, build_chat_client from ..rag.reranker import build_reranker from ..rag.retriever import RetrievalRequestError, Retriever from ..rag.service import DocumentService from .endpoints import router from .health import router as health_router from .ratelimit import SlidingWindowLimiter, client_identity logger = logging.getLogger(__name__) _STATIC_DIR = Path(__file__).parent / "static" DESCRIPTION = """ Document search and evidence-grounded question answering over a single local corpus. **Supported formats**: .txt, .md, .rst (plain text), .pdf (text layer only). **Retrieval modes**: `lexical` (BM25) always works; `vector` and `hybrid` need an embedding provider. Every response reports the mode that actually ran. **Answers** are generated only when a generation provider is configured and are checked so that every citation resolves to a passage that was in the model's context. Without a provider the API returns the supporting excerpts instead. **Access control**: when `API_KEY` is set every `/api/v1` route requires the `X-API-Key` header. Without it the server runs in local mode with no authentication, which is only appropriate on a trusted machine. There is no multi-tenant isolation: one process serves one corpus. """ def build_services(settings: Settings): service = DocumentService.from_settings(settings) chat = build_chat_client(settings) reranker = build_reranker(settings, chat) retriever = Retriever(service, settings, reranker) generator = Generator(retriever, settings, chat) return service, generator def create_app(settings: Optional[Settings] = None) -> FastAPI: settings = settings or get_settings() logging.basicConfig( level=getattr(logging, settings.log_level), format="%(asctime)s %(levelname)s %(name)s %(message)s", ) @asynccontextmanager async def lifespan(app: FastAPI): app.state.settings = settings app.state.service = None app.state.generator = None app.state.startup_error = None app.state.seeded_doc_ids = [] app.state.seed_errors = [] try: service, generator = build_services(settings) app.state.service = service app.state.generator = generator if settings.demo_seed_dir: seeded = await service.seed_directory(Path(settings.demo_seed_dir)) app.state.seeded_doc_ids = seeded.doc_ids app.state.seed_errors = seeded.errors for error in seeded.errors: logger.error("Seed document skipped: %s", error) logger.info( "Seeded %d document(s) from %s", len(seeded.doc_ids), settings.demo_seed_dir ) metrics.initialize_metrics( settings.app_version, settings.app_env, settings.resolved_embedding_provider, settings.resolved_generation_provider, ) logger.info( "Started (storage=%s, embeddings=%s, generation=%s, reranker=%s, auth=%s)", settings.storage_mode, settings.resolved_embedding_provider, settings.resolved_generation_provider, settings.reranker_mode, "api_key" if settings.api_key else "none", ) except IndexCompatibilityError as exc: app.state.startup_error = str(exc) logger.error("Startup failed: %s", exc) except Exception as exc: app.state.startup_error = f"{type(exc).__name__}: {exc}" logger.exception("Startup failed") yield if app.state.service is not None: app.state.service.close() app = FastAPI( title=settings.app_name, version=settings.app_version, description=DESCRIPTION, lifespan=lifespan, docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json", openapi_tags=[ {"name": "documents", "description": "Upload, inspect, replace, and delete documents"}, {"name": "search", "description": "Lexical, vector, and hybrid retrieval"}, {"name": "query", "description": "Grounded question answering"}, {"name": "evaluation", "description": "Optional LLM-based answer rating"}, {"name": "health", "description": "Liveness, readiness, statistics"}, ], ) app.state.settings = settings origins = settings.cors_origin_list if origins: app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials="*" not in origins, allow_methods=["GET", "POST", "DELETE"], allow_headers=["Content-Type", "X-API-Key"], ) app.include_router(router, prefix="/api/v1") app.include_router(health_router) limiter = SlidingWindowLimiter( settings.rate_limit_per_minute, settings.rate_limit_global_per_minute ) app.state.limiter = limiter if limiter.enabled: @app.middleware("http") async def rate_limit(request: Request, call_next): if request.url.path.startswith("/api/"): retry_after = limiter.check(client_identity(request, settings)) if retry_after is not None: return JSONResponse( status_code=429, content={ "error": "Rate limit exceeded; retry later", "code": "rate_limited", "request_id": getattr(request.state, "request_id", None), }, headers={"Retry-After": str(retry_after)}, ) return await call_next(request) if settings.metrics_enabled and metrics.AVAILABLE: from prometheus_client import CONTENT_TYPE_LATEST, generate_latest @app.get("/metrics", include_in_schema=False) async def metrics_endpoint() -> Response: return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST) @app.middleware("http") async def request_context(request: Request, call_next): request_id = uuid.uuid4().hex[:12] request.state.request_id = request_id start = time.perf_counter() if metrics.AVAILABLE: metrics.active_requests.inc() status_code = 500 try: response = await call_next(request) status_code = response.status_code response.headers["X-Request-ID"] = request_id return response finally: if metrics.AVAILABLE: metrics.active_requests.dec() route = request.scope.get("route") template = getattr(route, "path", None) or "unmatched" metrics.record_request( request.method, template, status_code, time.perf_counter() - start ) # -- error handling: preserve status codes, sanitize details ------------- def _error(request: Request, status: int, message: str, code: Optional[str] = None): return JSONResponse( status_code=status, content={ "error": message, "code": code, "request_id": getattr(request.state, "request_id", None), }, ) @app.exception_handler(IngestionError) async def _ingestion_error(request: Request, exc: IngestionError): return _error(request, exc.status, exc.message, exc.code) @app.exception_handler(NotFoundError) async def _not_found(request: Request, exc: NotFoundError): return _error(request, 404, f"Document not found: {exc}", "not_found") @app.exception_handler(RetrievalRequestError) async def _retrieval_error(request: Request, exc: RetrievalRequestError): return _error(request, exc.status, str(exc), "invalid_request") @app.exception_handler(ProviderError) async def _provider_error(request: Request, exc: ProviderError): return _error(request, 502, f"Generation provider error: {exc}", "provider_error") @app.exception_handler(HTTPException) async def _http_error(request: Request, exc: HTTPException): detail = exc.detail if isinstance(exc.detail, str) else None payload = { "error": detail or "Request failed", "code": None, "request_id": getattr(request.state, "request_id", None), } if not isinstance(exc.detail, str): payload["detail"] = exc.detail return JSONResponse(status_code=exc.status_code, content=payload, headers=exc.headers) @app.exception_handler(RequestValidationError) async def _validation_error(request: Request, exc: RequestValidationError): return JSONResponse( status_code=422, content={ "error": "Validation failed", "code": "validation_error", "detail": exc.errors(), "request_id": getattr(request.state, "request_id", None), }, ) @app.exception_handler(Exception) async def _unhandled(request: Request, exc: Exception): logger.exception("Unhandled error (request %s)", getattr(request.state, "request_id", "?")) message = ( f"{type(exc).__name__}: {exc}" if settings.is_development else "Internal server error" ) return _error(request, 500, message, "internal_error") # -- root and review UI ------------------------------------------------------ @app.get("/", tags=["health"], summary="Service information") async def root(): return { "name": settings.app_name, "version": settings.app_version, "docs": "/docs", "ui": "/ui", "health": "/health", "ready": "/ready", } @app.get("/ui", include_in_schema=False) async def review_ui(): page = _STATIC_DIR / "index.html" if not page.is_file(): raise HTTPException(status_code=404, detail="Review UI not installed") return FileResponse(page, headers={"Cache-Control": "no-cache"}) def custom_openapi(): if app.openapi_schema: return app.openapi_schema from fastapi.openapi.utils import get_openapi schema = get_openapi( title=app.title, version=app.version, description=app.description, routes=app.routes ) schema.setdefault("components", {})["securitySchemes"] = { "ApiKeyAuth": { "type": "apiKey", "in": "header", "name": "X-API-Key", "description": "Required on /api/v1 routes only when API_KEY is configured.", } } schema["servers"] = [{"url": "/", "description": "This server"}] app.openapi_schema = schema return schema app.openapi = custom_openapi # type: ignore[method-assign] return app app = create_app()