Spaces:
Running
Running
File size: 6,194 Bytes
0157ac7 f3220aa 0157ac7 f3220aa 0157ac7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | """FastAPI application factory and configuration."""
import traceback
from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI, Request
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from loguru import logger
from starlette.types import Receive, Scope, Send
from config.logging_config import configure_logging
from config.settings import get_settings
from providers.exceptions import ProviderError
from .admin import router as admin_router
from .routes import router
from .runtime import AppRuntime, startup_failure_message
from .validation_log import summarize_request_validation_body
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager."""
runtime = AppRuntime.for_app(app, settings=get_settings())
await runtime.startup()
yield
await runtime.shutdown()
class GracefulLifespanApp:
"""ASGI wrapper that reports startup failures without Starlette tracebacks."""
def __init__(self, app: FastAPI):
self.app = app
def __getattr__(self, name: str) -> Any:
return getattr(self.app, name)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "lifespan":
await self.app(scope, receive, send)
return
await self._lifespan(receive, send)
async def _lifespan(self, receive: Receive, send: Send) -> None:
settings = get_settings()
runtime = AppRuntime.for_app(self.app, settings=settings)
startup_complete = False
while True:
message = await receive()
if message["type"] == "lifespan.startup":
try:
await runtime.startup()
except Exception as exc:
await send(
{
"type": "lifespan.startup.failed",
"message": startup_failure_message(settings, exc),
}
)
return
startup_complete = True
await send({"type": "lifespan.startup.complete"})
continue
if message["type"] == "lifespan.shutdown":
if startup_complete:
try:
await runtime.shutdown()
except Exception as exc:
logger.error("Shutdown failed: exc_type={}", type(exc).__name__)
await send({"type": "lifespan.shutdown.failed", "message": ""})
return
await send({"type": "lifespan.shutdown.complete"})
return
def create_app(*, lifespan_enabled: bool = True) -> FastAPI:
"""Create and configure the FastAPI application."""
settings = get_settings()
configure_logging(
settings.log_file, verbose_third_party=settings.log_raw_api_payloads
)
app_kwargs: dict[str, Any] = {
"title": "Claude Code Proxy",
"version": "2.0.0",
}
if lifespan_enabled:
app_kwargs["lifespan"] = lifespan
app = FastAPI(**app_kwargs)
# Register routes
app.include_router(router)
app.include_router(admin_router)
# Exception handlers
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
"""Log request shape for 422 debugging without content values."""
body: Any
try:
body = await request.json()
except Exception as e:
body = {"_json_error": type(e).__name__}
message_summary, tool_names = summarize_request_validation_body(body)
logger.debug(
"Request validation failed: path={} query={} error_locs={} error_types={} message_summary={} tool_names={}",
request.url.path,
str(request.url.query),
[list(error.get("loc", ())) for error in exc.errors()],
[str(error.get("type", "")) for error in exc.errors()],
message_summary,
tool_names,
)
return await request_validation_exception_handler(request, exc)
@app.exception_handler(ProviderError)
async def provider_error_handler(request: Request, exc: ProviderError):
"""Handle provider-specific errors and return Anthropic format."""
err_settings = get_settings()
if err_settings.log_api_error_tracebacks:
logger.error(
"Provider Error: error_type={} status_code={} message={}",
exc.error_type,
exc.status_code,
exc.message,
)
else:
logger.error(
"Provider Error: error_type={} status_code={}",
exc.error_type,
exc.status_code,
)
return JSONResponse(
status_code=exc.status_code,
content=exc.to_anthropic_format(),
)
@app.exception_handler(Exception)
async def general_error_handler(request: Request, exc: Exception):
"""Handle general errors and return Anthropic format."""
settings = get_settings()
if settings.log_api_error_tracebacks:
logger.error("General Error: {}", exc)
logger.error(traceback.format_exc())
else:
logger.error(
"General Error: path={} method={} exc_type={}",
request.url.path,
request.method,
type(exc).__name__,
)
return JSONResponse(
status_code=500,
content={
"type": "error",
"error": {
"type": "api_error",
"message": "An unexpected error occurred.",
},
},
)
return app
def create_asgi_app() -> GracefulLifespanApp:
"""Create the server ASGI app with graceful lifespan failure reporting."""
return GracefulLifespanApp(create_app(lifespan_enabled=False))
|