Metadata-Version: 2.5
Name: neva-faststream
Version: 0.2.0
Summary: FastStream integration for the Neva framework.
Requires-Python: >=3.12
Requires-Dist: faststream>=0.6.6
Requires-Dist: python-neva>=3.5.0
Description-Content-Type: text/markdown

# neva-faststream

FastStream integration for the [Neva](https://pypi.org/project/python-neva/)
framework — the messaging counterpart to
[`neva-fastapi`](https://pypi.org/project/neva-fastapi/).

It marries [FastStream](https://faststream.ag2.ai/) brokers and subscribers to
neva's dishka-based dependency-injection container: a subscriber resolves
injected services the same way a route does, each message gets its own DI
scope, and broker lifecycle is driven by neva service providers.

## Usage

```python
# src/apps/worker.py
from faststream.rabbit import RabbitBroker
from neva.faststream import App, Inject

from src.settings import MainSettings

broker = RabbitBroker(MainSettings().rabbitmq.url_string)


@broker.subscriber("documents")
async def handle(body: dict, documents: Inject[DocumentService]) -> None:
    await documents.process(body)


app = App(broker, config_path="src/config")
```

```bash
faststream run src.apps.worker:app
```

`Inject[T]` resolves from a container scoped to the message being consumed, so
a `scoped` binding yields one instance per message. The facades work inside a
subscriber too — `App.make`, `DB`, `Log`, `Event` all reach that same scope,
not the application container.

The broker is bound into the container, so anything that publishes can inject
it rather than reach for a global:

```python
async def publish(broker: Inject[RabbitBroker]) -> None:
    await broker.publish(payload, "documents")
```

A single-broker app also binds its broker as `BrokerUsecase`. With several
brokers that interface would resolve to whichever was bound last, so only the
concrete broker types are bound and injection must name one.

### Service providers

`App.register` takes a neva `ServiceProvider`, and providers declared in the
`providers` config namespace are picked up as usual. A provider implementing
`lifespan()` is entered on startup and exited on shutdown, around the broker's
own lifecycle.

```python
app = App(broker, config_path="src/config")
_ = app.register(DocumentServiceProvider)
```

Pass `lifespan=` to `App` for startup work that isn't a provider's; it runs
inside the neva application's lifespan, so the container and facades are live.

### Injecting the message

`Inject[StreamMessage]` gives the subscriber the message being consumed.
Broker-specific message classes are deliberately not declared here — the
middleware puts one in the scope, but only a consumer knows which broker it
runs on, so `Inject[RabbitMessage]` needs a `from_context` of its own:

```python
class WorkerServiceProvider(ServiceProvider):
    @override
    def register(self) -> Result[Self, str]:
        self.from_context(RabbitMessage, scope=Scope.REQUEST)
        return Ok(self)
```

### Turning auto-injection off

Every subscriber is wrapped so `Inject` parameters resolve without decorating
each one. Pass `auto_inject=False` to opt out and decorate explicitly:

```python
from neva.faststream import inject

@broker.subscriber("documents")
@inject
async def handle(body: dict, documents: Inject[DocumentService]) -> None: ...
```

## Install

```bash
uv add neva-faststream
```

Broker drivers are FastStream's own extras and are not pulled in: install the
one you use, e.g. `uv add "faststream[rabbit]"`. Nothing in this package imports
a broker module, so it stays agnostic over which you pick.

## Layout

`neva` is a **namespace package** (no top-level `neva/__init__.py`); this repo
owns `neva/faststream/` and shares the `neva.*` namespace with `python-neva`.

## Develop

```bash
uv sync          # install/refresh deps
poe lint         # ruff check
poe fmt          # ruff format
poe tc           # pyrefly check
poe test         # pytest
poe test-cov     # pytest with coverage
```

`asyncio_mode = "auto"` is set, so async tests need no `@pytest.mark.asyncio`.

## Contributing

This repo follows the same conventions as the rest of the Neva ecosystem.

**Commits** use [Conventional Commits](https://www.conventionalcommits.org/)
with [gitmoji](https://gitmoji.dev/) prefixes, enforced by
[`cz_gitmoji`](https://github.com/ljnsn/cz-conventional-gitmoji). Commitizen is
provided as a dev dependency — run `cz commit` for the guided wizard, or format
manually as `:gitmoji: type(scope): subject`.

**Releases** are cut with commitizen from this repo's root:

```bash
cz bump                          # bump version in pyproject, write CHANGELOG, tag v<version>
git push --follow-tags origin main
```

`cz bump` derives the level (major/minor/patch) from the commits since the last
tag, updates `CHANGELOG.md`, and runs `scripts/retag-with-changelog.sh` to
rewrite the new tag with the rendered changelog as its annotation.
