modern-di-faststream

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.

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)

📦 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.

Metadata

Release files for modern-di-faststream 4.2.2

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for modern-di-faststream 4.2.2
File	Size	Uploaded
modern_di_faststream-4.2.2.tar.gz	6.3 kB	Oct 17, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for modern-di-faststream 4.2.2
File	Interpreter	ABI	Platform	Reset
modern_di_faststream-4.2.2-py3-none-any.whl 6.2 kB Oct 17, 2026	Python 3	none	any	Details

Total release size: 12.6 kB

Release files / modern_di_faststream-4.2.2.tar.gz

Download URL	modern_di_faststream-4.2.2.tar.gz
Size	6.3 kB
Tags	Source
SHA-256 checksum How to use checksums	`f66bc5149d2fa75b59094e6dd92f435b21e8b54aaa919c095d3de0f27c1c4892`
BLAKE2b-256 checksum How to use checksums	`cbace99babdd53b239f28e3280c8320aebd34d9131904ce11d54d8ff4e4019bf`
Upload date	Oct 17, 2026
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.2.2-py3-none-any.whl

Download URL	modern_di_faststream-4.2.2-py3-none-any.whl
Size	6.2 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`42d16a57c5adcece0543f1df182b71a13bc96bba72382ffaef43dde1a6ad200b`
BLAKE2b-256 checksum How to use checksums	`6a17045c0741ac046e140f6283138187a0920632cc2d7e95e532344f425da854`
Upload date	Oct 17, 2026
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}`