modern-di-faststream

Modern-DI integration for FastStream.

Installation

uv add modern-di-faststream      # or: pip install modern-di-faststream

Usage

setup_di registers the container and installs a broker middleware that builds a per-message child container; 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], validate=True)
setup_di(app, container)


@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

• setup_di(app, container) — stores the container in the app context, registers a shutdown hook, and adds the DI middleware to the broker
• FromDI(dependency, *, use_cache=True, cast=False) — FastStream Depends that resolves a provider (or type) from the per-message child container
• fetch_di_container(app) — returns the app-scoped container from the app context
• faststream_message_provider — ContextProvider for the current faststream.StreamMessage

📦 PyPI

📝 License

Part of modern-python

Browse the full list of templates and libraries in modern-python — see the org profile for the categorized index.