modern-di integration for FastStream.
Full guide: FastStream integration docs
Usage example: examples/
Installation
uv add modern-di-faststream # or: pip install modern-di-faststream
Usage
setup_di registers the root container and installs a broker middleware that builds a request container per message; FromDI resolves a provider (or type) into a subscriber parameter.
import dataclasses
import faststream
from faststream.nats import NatsBroker
from modern_di import Container, Group, Scope, providers
from modern_di_faststream import FromDI, setup_di
@dataclasses.dataclass(kw_only=True)
class Settings:
debug: bool = True
@dataclasses.dataclass(kw_only=True)
class GreetingHandler:
settings: Settings # auto-injected by type
class Dependencies(Group):
settings = providers.Factory(scope=Scope.APP, creator=Settings)
handler = providers.Factory(scope=Scope.REQUEST, creator=GreetingHandler)
broker = NatsBroker()
app = faststream.FastStream(broker)
container = Container(groups=[Dependencies])
setup_di(app, container)
container.validate() # optional fail-fast; must come after setup_di registers its providers
@broker.subscriber("greetings")
async def handle(name: str, handler: GreetingHandler = FromDI(Dependencies.handler)) -> None:
print(name, handler.settings.debug)
The current StreamMessage is resolvable within DI via the pre-built faststream_message_provider context provider.
Sharing a container
By default the app opens the container on startup and closes it after shutdown, which runs every APP-scoped finalizer. The close runs after every on_shutdown and after_shutdown hook, including hooks registered after setup_di, so they can still resolve APP-scoped dependencies. The teardown of a FastStream(lifespan=...) context manager runs after the close, because FastStream exits it after all hooks. If that teardown needs the container, or the container outlives the app, or other apps share it, pass manage_lifespan=False and open and close it yourself:
container = Container(groups=[Dependencies])
setup_di(app, container, manage_lifespan=False)
async def main() -> None:
async with container:
await app.run()
When several apps share one container, pass manage_lifespan=False to every one of them and open and close the container yourself, as above. APP-scoped finalizers then run once, when your async with block exits. An app left at the default closes the container when it stops, while the other apps may still be handling work. See FastAPI and FastStream sharing a container.
Running a broker inside another framework's lifespan
setup_di takes a FastStream or AsgiFastStream app. To run a broker inside another framework's lifespan, such as FastAPI's, wrap it in FastStream(broker) and start and stop that app from the lifespan:
stream = faststream.FastStream(broker)
setup_di(stream, container, manage_lifespan=False)
@contextlib.asynccontextmanager
async def lifespan(_: fastapi.FastAPI) -> AsyncIterator[None]:
async with container:
await stream.start()
try:
yield
finally:
await stream.stop()
stream.start() runs the on_startup hooks, where setup_di installs its middleware. start() and stop() do not run a FastStream(lifespan=...). The shared container recipe has the full example.
FastStream's FastAPI routers, such as faststream.nats.fastapi.NatsRouter, need only modern-di-fastapi: its FromDI resolves in router subscribers.
API
| Symbol | Description |
|---|---|
setup_di(app, container, *, manage_lifespan=True) |
Stores the container in the app context and registers startup/shutdown lifecycle hooks. On startup it reopens the container and adds the DI middleware to every broker of the app, including one added via app.add_broker after setup_di or by an on_startup hook registered before setup_di. After the after_shutdown hooks run, it closes the container. With manage_lifespan=False it never opens or closes the container, and the middleware still builds and closes a request container per message. Raises RuntimeError at startup if the app has no broker by then. Returns the container |
FromDI(dependency, *, use_cache=True, cast=False) |
FastStream Depends that resolves a provider (or type) from the request container. Raises RuntimeError naming setup_di when a message reaches it without the middleware installed |
fetch_di_container(app) |
Returns the root container from the app context |
faststream_message_provider |
ContextProvider for the current faststream.StreamMessage (REQUEST scope) |
Links
Part of modern-python
Built on modern-di, a dependency-injection framework with an IoC container and scopes.
Browse the full list of templates and libraries in
modern-python; the org profile has the categorized index.
1 maintainer
Metadata
1 maintainer
Release files for modern-di-faststream 4.2.4
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| modern_di_faststream-4.2.4.tar.gz | 6.5 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| modern_di_faststream-4.2.4-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 12.9 kB
Release files / modern_di_faststream-4.2.4.tar.gz
| Download URL | modern_di_faststream-4.2.4.tar.gz |
|---|---|
| Size | 6.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
208d8dd095fee3d919369dae871806a36ee56c2d64a10057e25cf77b66906910
|
|
BLAKE2b-256 checksum How to use checksums |
647f13ccb7cd4c8e2c4c469070e82d93bc1eab9b328b229e89faf60b921cabfa
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.4 {"installer":{"name":"uv","version":"0.13.4","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / modern_di_faststream-4.2.4-py3-none-any.whl
| Download URL | modern_di_faststream-4.2.4-py3-none-any.whl |
|---|---|
| Size | 6.4 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
a4a40e48a1509afe5d3a52c2cfe8c6129bda81eff1764cb4abc31a5003e5e204
|
|
BLAKE2b-256 checksum How to use checksums |
23a3e2aed8e30ea786b1d99029b4d1f76ad5473541ef2ad10c880626baeecf4f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.4 {"installer":{"name":"uv","version":"0.13.4","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Security for modern-di-faststream 4.2.4
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
