FastAPI¶
FastAPI is the optional second interface to the engine described in Architecture: the same analyze/deploy/validate operations the CLI exposes, reachable over HTTP, with an automatically generated OpenAPI document. It is explicitly the secondary interface; the CLI is primary.
App construction and lifespan¶
src/test_data_workbench/api/main.py builds one FastAPI instance with
an @asynccontextmanager lifespan function, the current FastAPI-recommended
replacement for the older @app.on_event("startup") style:
@asynccontextmanager
async def lifespan(app: FastAPI):
from test_data_workbench.core.feature_flags import FeatureFlags
from test_data_workbench.core.defensive_manager import DefensiveGeneratorManager
app.state.feature_flags = FeatureFlags()
app.state.defensive_manager = DefensiveGeneratorManager(app.state.feature_flags)
yield
app = FastAPI(
title="V4 Adaptive Framework API",
version="4.0.0",
docs_url="/docs",
redoc_url="/redoc",
lifespan=lifespan,
)
app.state is where the process-lifetime objects live, one
FeatureFlags instance and one DefensiveGeneratorManager (see
Architecture), constructed once at startup and read by
request handlers through getattr(app.state, "...", None) rather than
re-created per request.
Cross-cutting middleware¶
Two middleware layers wrap every request. CORSMiddleware allows the local
frontend origins used in development; and a small custom middleware records
per-request timing onto a response header:
@app.middleware("http")
async def track_requests(request, call_next):
global request_count
request_count += 1
start_time = time.time()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.time() - start_time)
return response
The same middleware increments a module-level request counter, which the
/health endpoint later reports. This is process-local, in-memory state,
consistent with the rest of this API layer: it is a coordination surface for
a team working together, not a service designed for horizontal scaling
behind a load balancer.
Errors degrade instead of surfacing a raw 500¶
Two exception handlers replace FastAPI’s default error bodies with structured, actionable ones:
@app.exception_handler(404)
async def not_found_handler(request, exc):
return JSONResponse(status_code=404, content={
"error": "Endpoint not found",
"available_endpoints": [...],
"suggestion": "Check the API documentation at /docs for available endpoints",
})
The schema-analysis endpoint itself goes further and does not raise an HTTP
error at all when analysis fails; analyze_production_schema in
api/endpoints/schema.py catches the exception and returns a
SchemaAnalysisResponse(success=False, ...) with the error folded into
analysis_metadata. This is the same “degrade, never fail” instinct from
Design Principles applied to the HTTP layer: a client gets
a 200 with a machine-readable failure description it can branch on,
rather than parsing exception text out of a 500 body.
BackgroundTasks for the one thing that can wait¶
The schema-analysis endpoint schedules cache trimming after the response is already on its way:
background_tasks.add_task(cleanup_old_cache_entries)
This is FastAPI’s BackgroundTasks used for exactly what it is for: work
that must happen, but that the caller does not need to wait on. The
in-memory analysis_cache dict it trims is, again, process-local, so this
API layer is explicitly scoped to single-process, team-local use rather than
production request serving; a comment in the source notes Redis as the
production alternative that this codebase does not include, consistent with
the deliberately short dependency list described in Technology Stack.