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.