Spaces:
Runtime error
Runtime error
feat(web): add vanilla JS one-file console with CORS, PDF upload, and custom themeFeat/web console (#18)
bc2b07b unverified | import uuid | |
| import logging | |
| from fastapi import FastAPI, Request, Depends | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.exceptions import RequestValidationError | |
| from starlette.exceptions import HTTPException as StarletteHTTPException | |
| import app | |
| from transparent_fastapi import install | |
| from app.config import settings | |
| from app.routers import auth, hubs, documents | |
| from app.errors import AppError, error_response | |
| from app.deps import get_current_user | |
| from app.models import User | |
| from app.services.groq_service import AiUnavailable | |
| logger = logging.getLogger("uvicorn.error") | |
| def create_app() -> FastAPI: | |
| app = FastAPI( | |
| title="ContextQuery AI", | |
| version="0.1.0", | |
| docs_url="/docs", | |
| redoc_url=None, | |
| ) | |
| # Initialize transparent-fastapi for /metrics | |
| install(app) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=[ | |
| "http://localhost:3000", | |
| "http://127.0.0.1:3000", | |
| "http://localhost:8000", | |
| "http://127.0.0.1:8000", | |
| "http://localhost:5500", # <-- Add this (Live Server localhost) | |
| "http://127.0.0.1:5500", | |
| "null", # <--- THIS IS CRITICAL FOR LOCAL FILE TESTING | |
| ], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| expose_headers=["X-Request-ID", "Idempotent-Replayed", "Retry-After"], | |
| ) | |
| app.include_router(auth.router, prefix="/api/v1") | |
| app.include_router(hubs.router, prefix="/api/v1") | |
| app.include_router(documents.router, prefix="/api/v1") | |
| async def catch_unhandled(request: Request, call_next): | |
| try: | |
| return await call_next(request) | |
| except Exception: | |
| logger.exception("unhandled error", extra={"request_id": getattr(request.state, "request_id", "unknown")}) | |
| return error_response(request, 500, "INTERNAL_ERROR", "Unexpected server error.") | |
| async def add_request_id(request: Request, call_next): | |
| request.state.request_id = str(uuid.uuid4()) | |
| response = await call_next(request) | |
| response.headers["X-Request-ID"] = request.state.request_id | |
| return response | |
| async def app_error_handler(request: Request, exc: AppError): | |
| return error_response(request, exc.status_code, exc.code, exc.message, exc.details, exc.headers) | |
| async def validation_error_handler(request: Request, exc: RequestValidationError): | |
| details = [] | |
| for error in exc.errors(): | |
| loc = error.get("loc", []) | |
| field = ".".join(str(x) for x in loc) if loc else "unknown" | |
| details.append({ | |
| "field": field, | |
| "reason": error.get("type", "invalid"), | |
| "message": error.get("msg", "Invalid value") | |
| }) | |
| return error_response(request, 422, "VALIDATION_ERROR", "Request validation failed.", details=details) | |
| async def http_exception_handler(request: Request, exc: StarletteHTTPException): | |
| if exc.status_code == 404: | |
| return error_response(request, 404, "ROUTE_NOT_FOUND", "The requested route does not exist.") | |
| if exc.status_code == 405: | |
| return error_response(request, 405, "METHOD_NOT_ALLOWED", "Method not allowed for this route.") | |
| return error_response(request, exc.status_code, "HTTP_ERROR", str(exc.detail)) | |
| AI_ERRORS = { | |
| "rate_limited": (429, "AI_RATE_LIMITED", "The AI provider's free quota is used up. Try again later."), | |
| "timeout": (504, "AI_TIMEOUT", "The AI provider took too long to answer."), | |
| "provider": (502, "AI_PROVIDER_ERROR", "The AI provider returned an error."), | |
| } | |
| async def ai_unavailable_handler(request: Request, exc: AiUnavailable): | |
| status, code, message = AI_ERRORS[exc.kind] | |
| headers = {"Retry-After": "30"} if exc.kind == "rate_limited" else None | |
| return error_response(request, status, code, message, headers=headers) | |
| def healthz() -> dict[str, str]: | |
| return {"status": "ok"} | |
| def protected_route(current_user: User = Depends(get_current_user)): | |
| return {"message": "You are authenticated!", "user_id": str(current_user.id)} | |
| return app | |
| app = create_app() |