Metadata-Version: 2.4
Name: modern-di-faststream
Version: 4.2.1
Summary: modern-di integration for FastStream
Keywords: dependency-injection,di,ioc-container,modern-di,faststream,python,asyncio
Author: Artur Shiriev
Author-email: Artur Shiriev <me@shiriev.ru>
License-Expression: MIT
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Free Threading :: 2 - Beta
Classifier: Typing :: Typed
Classifier: Topic :: Software Development :: Libraries
Requires-Dist: faststream>=0.7,<0.8
Requires-Dist: modern-di>=4.2,<5
Requires-Python: >=3.11, <4
Project-URL: Homepage, https://modern-di.modern-python.org
Project-URL: Documentation, https://modern-di.modern-python.org/integrations/faststream/
Project-URL: Repository, https://github.com/modern-python/modern-di-faststream
Project-URL: Issues, https://github.com/modern-python/modern-di-faststream/issues
Project-URL: Changelog, https://github.com/modern-python/modern-di-faststream/releases
Description-Content-Type: text/markdown

<p align="center">
  <picture>
    <source media="(prefers-color-scheme: dark)"  srcset="https://raw.githubusercontent.com/modern-python/.github/main/brand/projects/modern-di-faststream/lockup-dark.svg">
    <source media="(prefers-color-scheme: light)" srcset="https://raw.githubusercontent.com/modern-python/.github/main/brand/projects/modern-di-faststream/lockup-light.svg">
    <img alt="modern-di-faststream" src="https://raw.githubusercontent.com/modern-python/.github/main/brand/projects/modern-di-faststream/lockup.png" width="420">
  </picture>
</p>

[![PyPI version](https://img.shields.io/pypi/v/modern-di-faststream.svg)](https://pypi.org/project/modern-di-faststream/)
[![Supported Python versions](https://img.shields.io/pypi/pyversions/modern-di-faststream.svg)](https://pypi.org/project/modern-di-faststream/)
[![Downloads](https://static.pepy.tech/badge/modern-di-faststream/month)](https://pepy.tech/projects/modern-di-faststream)
[![Coverage](https://img.shields.io/badge/coverage-100%25-brightgreen.svg)](https://github.com/modern-python/modern-di-faststream/actions/workflows/ci.yml)
[![CI](https://github.com/modern-python/modern-di-faststream/actions/workflows/ci.yml/badge.svg)](https://github.com/modern-python/modern-di-faststream/actions/workflows/ci.yml)
[![License](https://img.shields.io/github/license/modern-python/modern-di-faststream.svg)](https://github.com/modern-python/modern-di-faststream/blob/main/LICENSE)
[![GitHub stars](https://img.shields.io/github/stars/modern-python/modern-di-faststream)](https://github.com/modern-python/modern-di-faststream/stargazers)
[![uv](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/uv/main/assets/badge/v0.json)](https://github.com/astral-sh/uv)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
[![ty](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ty/main/assets/badge/v0.json)](https://github.com/astral-sh/ty)

[modern-di](https://github.com/modern-python/modern-di) integration for [FastStream](https://faststream.ag2.ai).

Full guide: [FastStream integration docs](https://modern-di.modern-python.org/integrations/faststream/)

Usage example: [examples/](https://github.com/modern-python/modern-di-faststream/tree/main/examples)

## Installation

```bash
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.

```python
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. 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:

```python
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](https://modern-di.modern-python.org/recipes/shared-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:

```python
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](https://modern-di.modern-python.org/recipes/shared-container/#starting-faststream-inside-the-fastapi-lifespan) 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](https://pypi.org/project/modern-di-faststream)

## 📝 [License](https://github.com/modern-python/modern-di-faststream/blob/main/LICENSE)

## Part of `modern-python`

Built on [`modern-di`](https://github.com/modern-python/modern-di), a dependency-injection framework with an IoC container and scopes.

Browse the full list of templates and libraries in
[`modern-python`](https://github.com/modern-python); the org profile has the categorized index.
