FastAPI

FastAPI#

Why FastAPI#

The engine, SchemaAnalyzer, RapidAdapter, GeneratorFactory, and DefensiveGeneratorManager, is reachable primarily through the tdw command registered in pyproject.toml under [project.scripts]. FastAPI gives that same engine a second, optional way in: a teammate’s script, a browser hitting /docs, or a CI job can call POST /v4/schema/analyze or POST /v4/deployment/rapid over HTTP without shelling out to the CLI.

It was picked over hand-rolling a Starlette app, or reaching for a WSGI framework, because it turns work already being done for another reason, writing Pydantic request and response models (see Pydantic), directly into two things for free: request validation, and an OpenAPI document that /docs (Swagger UI) and /redoc (ReDoc) render without a separate schema-writing step. Nothing about the API layer needed to be described twice.

The CLI remains primary. There is no console-script entry for the API in pyproject.toml; starting it is a deliberate, separate action (see App construction and lifespan), and the HTTP surface it exposes is thinner than the CLI in places this page describes plainly in Sharp edges and limits.

The idea underneath#

FastAPI is built on ASGI, the Asynchronous Server Gateway Interface. ASGI is the async successor to WSGI (PEP 3333), the older synchronous calling convention between a Python web application and its server: under WSGI, one worker (a thread or process) is occupied for the full duration of one request, so serving many slow, I/O-bound connections at once means provisioning many workers. Under ASGI, a single process runs an event loop that holds many in-flight requests as suspended coroutines, resuming each one exactly when its I/O (a network read, a database round trip) completes. Uvicorn, the ASGI server this project depends on (uvicorn>=0.24.0 in pyproject.toml), is what actually runs that event loop; FastAPI (via Starlette) is the application that speaks the ASGI protocol to it. The specification itself lives at the ASGI documentation.

The important nuance: this is concurrency, not parallelism. One event loop means one thread making progress on many requests by interleaving them at await points, not multiple requests literally executing Python bytecode at the same instant. A coroutine that never actually yields, because it is doing synchronous, blocking work instead of awaiting real I/O, blocks that entire loop, not just its own request. This matters concretely for this codebase; see Sharp edges and limits.

FastAPI’s other core idea is dependency injection: a path operation can declare a parameter as Depends(some_callable), and FastAPI resolves and calls it per request, commonly used for shared database sessions, authentication, or pagination parameters. It is worth naming here because it is one of FastAPI’s signature features, but it is also worth being precise that none of the six routers in this project use it. Grepping src/test_data_workbench/api/ for Depends turns up nothing; every handler here is self-contained, either calling straight into the core engine or building its response directly.

The third idea is the OpenAPI document as a generated contract rather than a hand-maintained one. FastAPI walks every path operation, its declared response_model, and the Pydantic field metadata behind it, and builds a JSON Schema-based document from that, per the OpenAPI Specification (formerly “Swagger”). Nobody edits that document by hand; it is a byproduct of type hints that were going to exist anyway for validation.

How it fits this project#

src/test_data_workbench/api/main.py builds one FastAPI instance and mounts six routers on it, one per concern. Two of those routers, schema_router and deployment_router, call directly into the same SchemaAnalyzer and RapidAdapter classes the CLI uses (see Architecture); the API is a second door onto the same engine, not a parallel implementation of it.

The API also shows up reflexively inside the project’s own demo-readiness tooling. core/demo_coordinator.py defines DemoCoordinator, which tests six features before a presentation, and one of them, "Live API Demo", is tested by _test_live_api: it imports app from test_data_workbench.api.main, spins up fastapi.testclient.TestClient(app), and asserts that GET / returns 200. The FastAPI app is, among other things, one of the things this project checks is alive before someone demos it.

App construction and lifespan#

main.py builds the app 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 two process-lifetime objects live: one FeatureFlags and one DefensiveGeneratorManager, each constructed once at startup, not per request. Handlers read them defensively, through getattr(app.state, "feature_flags", None) rather than a direct attribute access, so a handler still returns a sensible response even if it somehow runs before or after the lifespan context is active, for example under a test harness that does not enter app as a context manager.

Router organisation and versioned prefixes#

api/endpoints/ has one module per area, schema, generators, templates, validation, deployment, team, each exporting a bare APIRouter(). main.py mounts all six under a versioned prefix with a tag for OpenAPI grouping:

app.include_router(schema_router, prefix="/v4/schema", tags=["Schema Analysis"])
app.include_router(generator_router, prefix="/v4/generators", tags=["Generator Creation"])
app.include_router(template_router, prefix="/v4/templates", tags=["Team Templates"])
app.include_router(validation_router, prefix="/v4/validation", tags=["Quality Gates"])
app.include_router(deployment_router, prefix="/v4/deployment", tags=["Rapid Deployment"])
app.include_router(team_router, prefix="/v4/team", tags=["Team Coordination"])

This is what gives /docs its grouped, tagged Swagger UI with no manual OpenAPI authoring: FastAPI walks the routers, their path operations, and the tags attached at mount time to build the document. A handful of routes, /, /health, and /v4/team/status, are registered directly on app in main.py instead of through a router; the /v4/team/status one duplicates ground also covered by team_router’s own GET /dashboard, so team-related state is readable from two different paths.

Request and response models#

Every request body and response shape in the API layer is a Pydantic BaseModel defined in api/models.py: DatabaseConnection, SchemaAnalysisRequest/Response, GeneratorCreationRequest/ Response, DataGenerationRequest/Response, RapidDeploymentRequest/Response, TeamTemplate, ContributionValidationRequest/Response, and more. That boundary, Pydantic in, Pydantic out, dataclasses everywhere else in the codebase, is covered in depth in Pydantic; this page only notes the mechanics that are FastAPI’s rather than Pydantic’s.

main.py pulls all of them in with from .models import * rather than naming them individually. api/models.py defines no __all__, so that star import re-exports everything at module scope, including names models itself imported: BaseModel, ConfigDict, Field, Dict, List, Any, Optional, datetime, and EntityType. None of those happen to collide with anything else main.py defines, so it works, but it means main.py’s namespace holds several names it never asked for directly.

response_model= on a path operation is what ties a handler’s return value to a schema FastAPI validates the outgoing response against and documents in /docs:

@router.post("/analyze", response_model=SchemaAnalysisResponse)
async def analyze_production_schema(request: SchemaAnalysisRequest,
                                     background_tasks: BackgroundTasks):
    ...

Cross-cutting middleware#

Two middleware layers wrap every request. CORSMiddleware allows requests from two hardcoded local development origins:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:8080"],  # Add your frontend URLs
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

The inline comment, “Add your frontend URLs”, is accurate: this list is not read from configuration anywhere, so pointing a real deployed frontend at this API means editing this literal and redeploying.

A small custom middleware records per-request timing onto a response header and increments a module-level counter:

@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

request_count is what /health later reports as requests_served. Because ASGI runs on one event loop per worker process and this increment has no await between the read and the write, it is safe from interleaving within a single worker; it is not safe from being wrong if the API is ever run behind more than one worker process, since each process keeps its own independent counter.

Error handling: what a caller actually sees on failure#

Two custom exception handlers replace FastAPI’s default error bodies with structured 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",
    })

@app.exception_handler(500)
async def internal_error_handler(request, exc):
    ...
    return JSONResponse(status_code=500, content={
        "error": "Internal server error",
        "incident_id": f"inc_{int(time.time())}",
        ...
    })

The 404 handler does fire for a genuinely unmatched route (tests/test_api.py asserts this directly against GET /nonexistent/path), because Starlette’s routing raises an HTTPException(status_code=404) internally when nothing matches, and a handler registered for an integer status code intercepts HTTPException instances carrying that code. See Sharp edges and limits for why the 500 handler does not behave the same way for a genuine unhandled exception.

The schema-analysis and rapid-deployment endpoints do not lean on either handler at all. Both wrap their real work in a broad except Exception as e and return a normal 200 with success=False, folding the error into the payload instead of raising:

except Exception as e:
    logger.error(f"Schema analysis failed: {str(e)}")
    return SchemaAnalysisResponse(
        success=False,
        database_name="unknown",
        tables_found=0,
        ...
        analysis_metadata={"error": str(e), "fallback_mode": True, ...},
    )

This is the “Degrade, never fail” principle from Design Principles applied at the HTTP boundary: a caller gets a 200 with a machine-readable failure description it can branch on, rather than parsing exception text out of a 500 body. It also means a client cannot trust the HTTP status code alone to detect failure on these two routes; it has to inspect success in the body.

Background tasks for the one real deferred job#

BackgroundTasks shows up in two routers, both to defer cleanup work past the point the response has already been sent. The schema-analysis endpoint schedules cache trimming:

background_tasks.add_task(cleanup_old_cache_entries)

which keeps the in-memory analysis_cache dict at 50 entries or fewer. The rapid-deployment endpoint schedules something with a much longer horizon:

async def cleanup_deployment(deployment_id: str):
    await asyncio.sleep(3600)
    if deployment_id in active_deployments:
        del active_deployments[deployment_id]

asyncio.sleep does not block the event loop, so this does not stall other requests, but it does mean every rapid deployment leaves a coroutine parked for an hour, and there is no cap on how many of those can be outstanding at once. Both caches, analysis_cache and active_deployments, are plain process-local dictionaries with no persistence: a restart forgets everything in them, cleanup tasks included.

Testing the endpoints#

tests/test_api.py exercises every router through Starlette’s TestClient, using it as a context manager:

@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

The with block is not decorative. Entering TestClient as a context manager is what triggers the app’s lifespan startup, so app.state.feature_flags and app.state.defensive_manager exist before any test request is sent, and shutdown runs when the fixture tears down. A bare TestClient(app) without the with would skip both.

The suite covers the root and health endpoints, every router’s happy path, the 404 fallback response, and the X-Process-Time header the timing middleware adds. pyproject.toml’s dev extra pins httpx2 for this, with an inline comment noting it is “required by starlette.testclient (FastAPI TestClient) on current starlette”. The broader philosophy behind the test suite, including that API endpoints are one of the areas it covers on every push, is described in Testing and Quality.

Sharp edges and limits#

Most routers are thinner than the CLI. Only schema_router and deployment_router call into real engine classes (SchemaAnalyzer, RapidAdapter). generators.py’s POST /v4/generators/create does not call GeneratorFactory at all, its own comment says so directly, “# Implementation would use GeneratorFactory”, and instead returns counts derived from len(request.entity_types). validation.py’s POST /v4/validation/validate-contribution always returns valid=True regardless of the submitted content. team.py’s GET /dashboard returns a fixed active_members: 5. This is consistent with the API being explicitly secondary, but it means the OpenAPI schema documents endpoints that do not yet do the thing their name and description imply.

Async endpoints wrapping blocking, synchronous database calls. SchemaAnalyzer.analyze_production_schema in src/test_data_workbench/adaptation/schema_analyzer.py is async def, but it opens a plain synchronous SQLAlchemy engine (sa.create_engine, not an async engine) and its helper coroutines, _get_sample_values, _estimate_row_count, _detect_business_rules, call engine.connect() and conn.execute() (or pd.read_sql) with no await inside them and no offload via asyncio.to_thread or run_in_executor. Against an in-memory demo backend this is invisible; a handler is only as async as the work it awaits, and coroutines that never actually suspend are, in practice, synchronous functions wearing async syntax. Against a real database, every one of those calls blocks the single ASGI event loop for its full round-trip time, not just the request making it. The asyncio.gather(*tasks) over per-table analysis in the same function does not produce real concurrency here either, since none of the gathered coroutines actually yield control before returning; they run one after another, serially, each still holding the loop.

No authentication on any route. Nothing in api/ checks an API key, token, or credential; every route mounted on the app is reachable by anyone who can open a TCP connection to it. Combined with allow_credentials=True on a CORS policy that currently only allows two localhost origins, this is workable for local, single-team use and not something to expose past that without adding an auth layer first.

``@app.exception_handler(500)`` does not catch a real unhandled exception. Registering a handler against the integer status code 500 only intercepts an explicitly raised HTTPException(status_code=500, ...); Starlette routes genuinely unhandled exceptions through a separate ServerErrorMiddleware instead, which the status-code handler never sees. Grepping this codebase for raise HTTPException turns up exactly two call sites, both status_code=404 (in schema.py and deployment.py); nothing anywhere raises a 500. The structured incident_id error body internal_error_handler promises is therefore unreachable for the case it was written for, a genuine bug in a handler.

All server-side state is process-local memory. analysis_cache, active_deployments, and request_count/startup_time are plain module-level dictionaries and variables. None of it persists across a restart, and none of it is shared if the API is ever run with more than one uvicorn worker; each worker would report a different health check and cache different analyses.

The dev-runner entry point is inconsistent with the installed package layout. The if __name__ == "__main__": block at the bottom of main.py calls uvicorn.run("api.main:app", ...). That import string names a top-level api package; the actual installed layout is test_data_workbench.api.main (pyproject.toml packages src/test_data_workbench, and there is no other api directory in the repository). There is no test covering this block. The API is meant to be started with an explicit, correctly-qualified invocation such as uvicorn test_data_workbench.api.main:app, not by running this file directly and trusting the string inside it.

Learn more resources#

Official documentation#

Tutorials and blogs#

Videos#