FastAPI integration
FastAPI's dependency system is well suited to HTTP concerns such as authentication, request parsing, headers, and authorization. Clean IoC complements it with a compiled container for the application service graph. The integration owns one child scope for each complete HTTP request or WebSocket connection while application services remain independent of FastAPI.
Installation
Clean IoC V2 supports FastAPI 0.121 and newer.
When to use both
| Concern | Native FastAPI | FastAPI with Clean IoC |
|---|---|---|
| HTTP parameters, security, and authorization | Depends, Security, Header, Query, and related APIs |
Continue using FastAPI APIs |
| Route-level application service | Depends(provider_function) |
Resolve(Service) |
| Deep application dependencies | Nested provider functions or FastAPI annotations in application constructors | Standard constructor annotations compiled from registrations |
| Request reuse | Dependency callable cache, enabled by default | per_resolution or request-owned scoped components |
| Application singleton | Application lifespan state or another application-managed cache | lifespan="singleton" |
| Application graph validation | FastAPI validates the route dependency model | Clean IoC validates component selection, cycles, generic plans, decorators, and lifespan ownership |
FastAPI supports arbitrarily deep sub-dependencies and caches repeated dependency callables within a request. Clean IoC does not replace those capabilities. It provides a separate composition model for application services that need to remain portable across HTTP handlers, workers, CLIs, message consumers, and tests.
Resolve at the route boundary
At a route signature, Resolve occupies the same dependency-marker position as Depends:
# Native FastAPI provider
async def endpoint(service: Service = Depends(provide_service)):
return await service.run()
# Compiled Clean IoC component
async def endpoint(service: Service = Resolve(Service)):
return await service.run()
After install_fastapi(app, container), replacing a route-level application provider is normally a one-line change.
Resolve always calls resolve_async, so the selected plan may contain both synchronous and asynchronous activation.
Keep using native FastAPI dependencies for transport-specific values and security policies.
Minimal application
from fastapi import FastAPI
from clean_ioc import ContainerBuilder
from clean_ioc.ext.fastapi import Resolve, install_fastapi
class Repository:
pass
class Service:
def __init__(self, repository: Repository):
self.repository = repository
builder = ContainerBuilder()
builder.register(Repository, lifespan="scoped")
builder.register(Service)
container = builder.build()
app = FastAPI()
install_fastapi(app, container)
@app.get("/")
async def endpoint(service: Service = Resolve(Service)):
return {"repository": type(service.repository).__name__}
install_fastapi owns the container for the application lifespan and an ordinary child scope for the full ASGI
operation. Its lifecycle behavior is provided by the dependency-free
ASGI extension; FastAPI adds route dependency resolution, framework request values, and startup validation.
Deep application dependency chains
Deep dependency chains are valid in native FastAPI. Keeping the application classes independent of FastAPI, however, requires a provider function for each layer:
from typing import Annotated
from fastapi import Depends
async def provide_database_session() -> DatabaseSession:
return DatabaseSession()
async def provide_order_repository(
session: Annotated[DatabaseSession, Depends(provide_database_session)],
) -> OrderRepository:
return OrderRepository(session)
async def provide_unit_of_work(
repository: Annotated[OrderRepository, Depends(provide_order_repository)],
) -> UnitOfWork:
return UnitOfWork(repository)
async def provide_place_order(
unit_of_work: Annotated[UnitOfWork, Depends(provide_unit_of_work)],
) -> PlaceOrder:
return PlaceOrder(unit_of_work)
async def provide_order_endpoint_service(
place_order: Annotated[PlaceOrder, Depends(provide_place_order)],
) -> OrderEndpointService:
return OrderEndpointService(place_order)
FastAPI resolves this chain correctly and caches each provider result for the request. The provider layer becomes larger
as the application graph grows. Putting Depends directly in application constructors removes some providers, but makes
those classes depend on FastAPI.
With Clean IoC, the same graph is defined by ordinary constructors and the composition root:
builder = ContainerBuilder()
builder.register(DatabaseSession, lifespan="scoped")
builder.register(OrderRepository)
builder.register(UnitOfWork)
builder.register(PlaceOrder)
builder.register(OrderEndpointService)
container = builder.build()
install_fastapi(app, container)
@app.post("/orders")
async def create_order(service: OrderEndpointService = Resolve(OrderEndpointService)):
return await service.run()
OrderRepository, UnitOfWork, PlaceOrder, and OrderEndpointService use the default per_resolution lifespan. The
database session is shared by every consumer in the request scope. The same compiled entry point can be resolved from a
worker or CLI without recreating the FastAPI provider chain.
Lifecycle behavior
The integration uses pure ASGI middleware rather than a yield dependency to own scopes. This means:
scopedservices are shared within one request or WebSocket connection and isolated between operations;- singleton services belong to the application container;
- streaming responses and background work finish before request-owned resources are finalized;
- WebSocket resources live until the connection handler exits;
- exceptions still close the request scope;
- no dependency plan is compiled during a request.
Custom FastAPI lifespans continue to work. The middleware enters the container around FastAPI's own lifespan:
install_fastapi is the only lifecycle integration. It keeps request, streaming-response, WebSocket, background-work, and cleanup ownership inside one ASGI boundary.
Lifespans and application singletons
FastAPI's dependency cache reuses a provider result within one request. Application-wide resources with startup and shutdown work are normally managed through the FastAPI application lifespan:
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
client = HttpClient()
await client.open()
app.state.http_client = client
try:
yield
finally:
await client.close()
app = FastAPI(lifespan=lifespan)
def provide_http_client(request: Request) -> HttpClient:
return request.app.state.http_client
Clean IoC keeps the ownership policy with the component registration:
from contextlib import asynccontextmanager
@asynccontextmanager
async def http_client_factory():
client = HttpClient()
await client.open()
try:
yield client
finally:
await client.close()
builder.register(
HttpClient,
factory=http_client_factory,
lifespan="singleton",
)
| Clean IoC lifespan | Ownership in a FastAPI application |
|---|---|
transient |
New activation for each dependency edge |
per_resolution |
Shared within one top-level application-service resolution |
scoped |
Shared for the complete HTTP request or WebSocket connection |
singleton |
Shared by the application container and finalized at shutdown |
Singleton activation is lazy by default. Resolve a singleton from a custom application lifespan when eager initialization is required. The compiler rejects a singleton that directly or transitively captures request-scoped state.
Build and startup validation
The integration has two validation points:
ContainerBuilder.build()compiles every visible application component plan and rejects missing components, cycles, invalid generic specialization, decorator or pre-configuration errors, and captive lifespan paths.- During ASGI startup,
install_fastapiverifies that every route-levelResolveservice and filter exists in the frozen container.
Every Resolve dependency records its requested service type and component filter. Before FastAPI starts accepting traffic, the integration checks those requests against the frozen container.
import clean_ioc.component_filters as cf
@app.get("/primary")
async def primary(service: Service = Resolve(Service, filter=cf.with_name("primary"))):
return service
If the primary component does not exist, application startup raises FastAPIIntegrationError with the route and missing service. The first request is not used as validation.
FastAPI also performs early validation: it builds its native dependency model while routes are registered and validates the parameter declarations it owns. The distinction is scope. FastAPI validates the HTTP dependency tree; Clean IoC validates the separately registered application component graph and its ownership rules.
Request and response boundary values
Apply FastAPIBundle before build() when application components consume HTTP connection information or the framework-light header adapters:
from clean_ioc.ext.fastapi import (
FastAPIBundle,
RequestHeaderReader,
ResponseHeaderWriter,
)
class HeaderAwareService:
def __init__(self, headers: RequestHeaderReader, response_headers: ResponseHeaderWriter):
self.headers = headers
self.response_headers = response_headers
def run(self) -> str:
self.response_headers.write("X-Clean-IoC", "active")
return self.headers.read("X-Request-ID")
builder = ContainerBuilder()
builder.apply_bundle(FastAPIBundle())
builder.register(HeaderAwareService)
container = builder.build()
app = FastAPI()
install_fastapi(app, container)
No global dependencies=[Depends(...)] list is required. FastAPIBundle runs at most once per builder, even when
independently composed application bundles apply it more than once. It:
- applies
ASGIBundle, which declaresASGIConnection,ResponseHeaderWriter, and the scopedRequestHeaderReader; - declares additional slots for
HTTPConnection,Request, andWebSocket; - lets the middleware provide the current boundary values before resolution.
Application code can inject fastapi.Request or fastapi.WebSocket directly when framework coupling is intentional. Prefer RequestHeaderReader and ResponseHeaderWriter in portable application services.
WebSockets
Resolve uses Starlette's shared HTTPConnection boundary, so the same API works for HTTP and WebSocket routes:
from fastapi import WebSocket
@app.websocket("/events")
async def events(websocket: WebSocket, handler: EventHandler = Resolve(EventHandler)):
await websocket.accept()
await handler.run(websocket)
One Clean IoC scope is retained for the entire connection. ResponseHeaderWriter values written before websocket.accept() are added to the handshake headers.
Async resources
from contextlib import asynccontextmanager
import httpx
@asynccontextmanager
async def http_client_factory():
async with httpx.AsyncClient() as client:
yield client
builder = ContainerBuilder()
builder.register(httpx.AsyncClient, factory=http_client_factory, lifespan="singleton")
builder.register(ExternalApi, lifespan="scoped")
container = builder.build()
Application-owned async singletons are finalized at FastAPI shutdown. Request-owned async resources are finalized after the HTTP response, stream, background work, or WebSocket handler completes.
Testing overrides
Keep application construction injectable so a compiled overlay can become the test root without mutating the production container:
from fastapi.testclient import TestClient
from clean_ioc import Scope
def create_app(root_scope: Scope) -> FastAPI:
app = FastAPI()
install_fastapi(app, root_scope)
return app
test_builder = container.new_scope_builder()
test_builder.register(PaymentGateway, FakePaymentGateway)
test_scope = test_builder.build()
with TestClient(create_app(test_scope)) as client:
response = client.post("/checkout")
The installed application owns and closes the overlay. For request-specific values, use the slots installed by
FastAPIBundle. Ordinary request scopes reuse the existing plan; an overlay compiles a new plan.
Complete example
See examples/fastapi_clean_architecture for a framework-independent use case, scoped repository, singleton adapters, an audit decorator, startup validation, and endpoint resolution.