Usage with Starlette¶
Starlette has no dependency-injection system of its own, so modern-di-starlette
uses the @inject decorator with FromDI markers (there is no Depends).
setup_di composes the lifespan and installs middleware that opens a
per-connection child container automatically.
How to use¶
1. Install modern-di-starlette¶
2. Apply to your application¶
import dataclasses
import typing
from modern_di import Container, Group, Scope, providers
from modern_di_starlette import FromDI, inject, setup_di
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
@dataclasses.dataclass(kw_only=True, slots=True, frozen=True)
class Settings:
service_name: str = "catalog"
@dataclasses.dataclass(kw_only=True, slots=True)
class Report:
settings: Settings # APP-scoped, injected by type
def as_dict(self) -> dict[str, str]:
return {"service": self.settings.service_name}
class AppGroup(Group):
settings = providers.Factory(Settings, scope=Scope.APP, cache=True)
report = providers.Factory(Report, scope=Scope.REQUEST)
@inject
async def get_report(
request: Request,
report: typing.Annotated[Report, FromDI(Report)],
) -> JSONResponse:
return JSONResponse(report.as_dict())
app = Starlette(routes=[Route("/report", get_report)])
container = Container(groups=[AppGroup])
setup_di(app, container)
container.validate() # after setup_di — its connection providers are now registered
Deployment: mounted sub-apps and disabled lifespan
Starlette only opens the root container from the ASGI lifespan event.
A setup_di-wired app mounted as a sub-application
(app.mount("/sub", subapp)) never receives that event from its parent,
and deployments that disable lifespan (e.g. Mangum lifespan="off") skip
it too. Requests still succeed (the container is already open from
construction), but nothing ever closes it, so its finalizers never run at
shutdown. Call setup_di on the top-level served app, or close the
root yourself (await container.close_async()) at shutdown.
3. Scopes¶
See the scope hierarchy.
An HTTP request opens a Scope.REQUEST child container; a WebSocket connection
opens a Scope.SESSION one, built by the middleware before your handler runs
and kept open for the whole life of the connection.
Class-based endpoints¶
@inject works the same on the methods of an HTTPEndpoint or WebSocketEndpoint
subclass. Decorate the handler method, not the class. self and any arguments
Starlette passes after the connection are forwarded unchanged, so
WebSocketEndpoint.on_receive and on_disconnect inject too:
import typing
from modern_di_starlette import FromDI, inject
from starlette.applications import Starlette
from starlette.endpoints import HTTPEndpoint, WebSocketEndpoint
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route, WebSocketRoute
from starlette.websockets import WebSocket
class ReportEndpoint(HTTPEndpoint):
@inject
async def get(
self,
request: Request,
report: typing.Annotated[Report, FromDI(Report)],
) -> JSONResponse:
return JSONResponse(report.as_dict())
class EchoEndpoint(WebSocketEndpoint):
encoding = "text"
@inject
async def on_receive(
self,
websocket: WebSocket,
data: str,
settings: typing.Annotated[Settings, FromDI(Settings)],
) -> None:
await websocket.send_text(f"{settings.service_name}: {data}")
app = Starlette(routes=[Route("/report", ReportEndpoint), WebSocketRoute("/echo", EchoEndpoint)])
Requires modern-di-starlette 3.2.0 or later. Earlier versions raise a
TypeError on the first request to a decorated method.
Websockets¶
For per-message work within a websocket's Scope.SESSION container, open a
nested Scope.REQUEST child:
import typing
import modern_di
from modern_di import Scope, providers
from modern_di_starlette import FromDI, inject
from starlette.websockets import WebSocket
@inject
async def ws_handler(
websocket: WebSocket,
container: typing.Annotated[modern_di.Container, FromDI(providers.container_provider)],
) -> None:
await websocket.accept()
async for message in websocket.iter_text():
async with container.build_child_container(scope=Scope.REQUEST) as request_container:
... # resolve REQUEST-scoped providers for this message
Framework context objects¶
Framework-specific context objects like starlette.requests.Request and
starlette.websockets.WebSocket are automatically made available by the
integration. See Framework context objects
for how implicit and explicit resolution work.
The following context providers are available for import:
starlette_request_providerprovides the currentstarlette.requests.Request(REQUEST scope)starlette_websocket_providerprovides the currentstarlette.websockets.WebSocket(SESSION scope)
Implicit (type-based) usage¶
from starlette.requests import Request
from modern_di import Group, Scope, providers
def create_request_info(request: Request) -> dict[str, str]:
return {"method": request.method, "url": str(request.url)}
class AppGroup(Group):
request_info = providers.Factory(create_request_info, scope=Scope.REQUEST)
Explicit (provider-based) usage¶
import modern_di_starlette
from starlette.requests import Request
from modern_di import Group, Scope, providers
def create_request_info(request: Request) -> dict[str, str]:
return {"method": request.method, "url": str(request.url)}
class AppGroup(Group):
request_info = providers.Factory(
create_request_info,
scope=Scope.REQUEST,
kwargs={"request": modern_di_starlette.starlette_request_provider},
)
See also¶
- Testing with overrides: swap providers in your tests.
- Async SQLAlchemy: engine + session + repository through the request container.
- Lifecycle: finalizers and
close_async(). - Scopes: the APP → REQUEST lifetime model.
API¶
| Symbol | Description |
|---|---|
setup_di(app, container) |
Registers the container on app.state, composes the lifespan (opens/closes the container), and installs the middleware that builds a per-connection child container; returns the container. |
FromDI(dependency) |
Marker (used with @inject) that resolves a provider or type from the per-connection child container. |
inject |
Decorator for an async def handler(connection: Request | WebSocket, ...), a function endpoint or a method of an HTTPEndpoint / WebSocketEndpoint subclass; resolves its FromDI-annotated parameters. Raises RuntimeError naming setup_di when the connection did not pass through the middleware. |
fetch_di_container(app) |
Returns the root Container stored on app.state. |
starlette_request_provider |
ContextProvider for starlette.requests.Request (REQUEST scope), auto-registered. |
starlette_websocket_provider |
ContextProvider for starlette.websockets.WebSocket (SESSION scope), auto-registered. |