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.
API
| Symbol | Description |
|---|---|
setup_di(app, container) |
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. 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.0.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.0.0.tar.gz | 5.3 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| modern_di_faststream-4.0.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 10.9 kB
Release files / modern_di_faststream-4.0.0.tar.gz
| Download URL | modern_di_faststream-4.0.0.tar.gz |
|---|---|
| Size | 5.3 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
18b723d9053b223a8bca8490cc863eaea8ad62962cf713c84b8573d6a3e63a72
|
|
BLAKE2b-256 checksum How to use checksums |
650b224664e6b3a5f46a0cefe6eacbd013f16016ed85eb49235b6cb7324a0e69
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.12.24 {"installer":{"name":"uv","version":"0.12.24","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.0.0-py3-none-any.whl
| Download URL | modern_di_faststream-4.0.0-py3-none-any.whl |
|---|---|
| Size | 5.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
8185788aaae68aba6a66387665d0a75e69d18a308e88b3ec931fbaad114f54db
|
|
BLAKE2b-256 checksum How to use checksums |
8e14ad762756802d5c951a0dfa5e77280e8b2a46a10536ddc0a1b91b4ed69d10
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.12.24 {"installer":{"name":"uv","version":"0.12.24","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}
|
