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. When 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, exactly one of them should own its lifespan, or the first app to stop closes the container for the rest.
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 shutdown 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) |
📦 PyPI
📝 License
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.1.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| modern_di_faststream-4.1.0.tar.gz | 5.9 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| modern_di_faststream-4.1.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 11.8 kB
Release files / modern_di_faststream-4.1.0.tar.gz
| Download URL | modern_di_faststream-4.1.0.tar.gz |
|---|---|
| Size | 5.9 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
62bcba52064ab88db7ce69e1cff6541791ecfdcb31141d581181e361cf9aa4a3
|
|
BLAKE2b-256 checksum How to use checksums |
43ae5b03155c62a765feab6cf9264ca6b7305763b916303bc78d67476c5b2714
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","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.1.0-py3-none-any.whl
| Download URL | modern_di_faststream-4.1.0-py3-none-any.whl |
|---|---|
| Size | 5.9 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
1516dcc8f5a31c7e80d4ecd912f0d0db4413b870de8d856a00c08a69ff98a605
|
|
BLAKE2b-256 checksum How to use checksums |
e6c8d4ac614795aeb573aa19453f5e9ab5be00283f77b439be8d06d9cca0d8c5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","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}
|
