Spaces:
Configuration error
Configuration error
| """FastAPI transport for the stateless numeric inference domain.""" | |
| import hmac | |
| import os | |
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.exceptions import RequestValidationError | |
| from fastapi.responses import JSONResponse | |
| from .admission import admit, recognize_semaphore | |
| from .model_registry import ModelRegistry | |
| from .numeric_routes import API_VERSION, MAX_BODY_BYTES, register_numeric_routes | |
| from .service import NumericInferenceService | |
| PUBLIC_PATHS = {"/health/live", "/health/ready"} | |
| def require_api_key() -> str: | |
| """Fail closed when this process is bound off loopback without a key.""" | |
| key = os.getenv("AXIS_API_KEY", "").strip() | |
| host = os.getenv("AXIS_BIND_HOST", "127.0.0.1").strip() or "127.0.0.1" | |
| if host not in {"127.0.0.1", "::1", "localhost"} and not key: | |
| raise SystemExit("AXIS_API_KEY is required when AXIS_BIND_HOST is not loopback") | |
| return key | |
| def create_app(service: NumericInferenceService, registry: ModelRegistry) -> FastAPI: | |
| api_key = require_api_key() | |
| slots = recognize_semaphore() | |
| app = FastAPI(title="AXIS Numeric Inference", version=API_VERSION) | |
| async def request_guard(request: Request, call_next): | |
| request_id = request.headers.get("x-request-id", "")[:128] or "server-generated" | |
| if api_key and request.url.path not in PUBLIC_PATHS: | |
| supplied = request.headers.get("x-api-key", "") or request.headers.get("authorization", "") | |
| if supplied.startswith("Bearer "): | |
| supplied = supplied[7:] | |
| if not hmac.compare_digest(supplied, api_key): | |
| return JSONResponse( | |
| status_code=401, | |
| content={"detail": "authentication_required"}, | |
| headers={"x-request-id": request_id}, | |
| ) | |
| declared = request.headers.get("content-length") | |
| if declared and declared.isdigit() and int(declared) > MAX_BODY_BYTES: | |
| return JSONResponse( | |
| status_code=413, | |
| content={"detail": "request_body_too_large"}, | |
| headers={"x-request-id": request_id}, | |
| ) | |
| response = await admit(request, call_next, slots) | |
| response.headers["x-request-id"] = request_id | |
| response.headers["x-api-version"] = API_VERSION | |
| return response | |
| async def validation_error(_: Request, __: RequestValidationError) -> JSONResponse: | |
| return JSONResponse(status_code=422, content={"detail": "invalid_numeric_request"}) | |
| def live() -> dict[str, str]: | |
| return {"status": "live"} | |
| def ready() -> dict[str, str]: | |
| if not registry.ready: | |
| raise HTTPException(status_code=503, detail="numeric_model_unavailable") | |
| return {"status": "ready", "modelVersion": registry.version} | |
| def lines_model() -> dict[str, str | bool]: | |
| # This process has no line recognizer. A 200 keeps the gateway probe off a 404. | |
| return {"ready": False, "status": "unavailable", "version": "", "apiVersion": API_VERSION} | |
| register_numeric_routes(app, service, registry) | |
| return app | |