Metadata-Version: 2.5
Name: neva-faststream
Version: 0.4.1
Summary: FastStream integration for the Neva framework.
Requires-Python: >=3.12
Requires-Dist: faststream>=0.6.6
Requires-Dist: python-neva>=5.3.0
Provides-Extra: testing
Requires-Dist: python-neva[testing]; extra == 'testing'
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 — the `App` facade's `make`, `DB`, `Log`, `Event` all reach that
same scope, not the application container.

The broker is bound into the container, so anything that publishes reaches it
from there rather than from a global — `Inject[T]` in a subscriber, a
constructor parameter in a service:

```python
@broker.subscriber("documents")
async def handle(body: DocumentPayload, publisher: Inject[RabbitBroker]) -> None:
    await publisher.publish(body.receipt, "receipts")


class NotifyDocument:
    def __init__(self, broker: RabbitBroker) -> None:
        self._broker = broker
```

`Inject` only means anything where FastStream does the calling; in a service it
is a missing positional argument.

A single-broker app also binds its broker as `BrokerUsecase`. With several
brokers that interface would resolve to whichever was bound last, so it is left
unbound and injection must name a concrete type.

### Naming a broker

A type only says which broker is meant while there is one of each kind. Two
`RabbitBroker` instances share a type, so neither is bound under it — name them
instead, and anything choosing a broker from config chooses it from the
`BrokerRegistry`:

```python
app = App(main_broker, named={"analytics": analytics_broker}, config_path="src/config")
```

The first positional broker is also named `default`. Named brokers are consumed
and wired like positional ones, so a broker is never passed twice.

```python
registry = app.application.make(BrokerRegistry).unwrap()
registry.get("analytics")  # Ok(analytics_broker)
registry.get()  # Ok(main_broker) — the default
registry.get("reports")  # Err, naming the brokers there are
```

`neva-queue` reads this registry: a queue connection names the broker it
publishes to, and two connections can reach two brokers.

### Every broker goes to the constructor

The set of brokers is fixed when the `App` is built. `app.add_broker(...)` and
the deprecated `app.set_broker(...)` raise `RuntimeError` rather than take a
late one: the container has already been built, so the broker would get no type
binding, no registry name, no container middleware and no injection. Left to
run it fails at message time with a missing-argument error on the subscriber's
injected parameter — a handler that looks correctly written.

One broker also belongs to one `App`. Building a second `App` over the same
broker is refused, because both container middlewares would run and the inner
one would open a message scope inside the outer one's.

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

`app.lifespan()` boots the application on its own — providers, listeners, the
container and the facades — but neither connects nor starts the brokers, so
publishing inside it raises `IncorrectState`. Enter the broker alongside it:

```python
async with app.lifespan(), broker:
    await broker.publish(payload, "documents")
```

It is the inner layer of the composed lifespan, so a `lifespan=` passed to
`App` does not run there; `faststream run` enters the composed one.

### Application metadata

The AsyncAPI document's title, version and description come from the `app`
config namespace rather than the constructor:

```python
# src/config/app.py
config = {
    "title": "Billing worker",
    "version": "2.1.0",
    "description": "Consumes billing events.",
}
```

`MessagingAppConfig` declares those three keys. The `app` namespace is
shared, and each package declares only what it reads — the core owns `key`,
`previous_keys` and `providers`. A `TypedDict` is closed, so annotate a real
`config/app.py` with a shape that inherits from each package's:

```python
from neva.arch import AppConfig
from neva.faststream import MessagingAppConfig


class WorkerAppConfig(MessagingAppConfig, AppConfig): ...


config: WorkerAppConfig = {"title": "Billing worker", "providers": [...]}
```

Unset keys fall back to `Neva Application` and `0.1.0`, and an unset
description is left out of the document.

For what config does not carry — a license, a contact, tags — pass a
specification of your own, which is then used as it stands:

```python
from faststream.specification import AsyncAPI

app = App(broker, specification=AsyncAPI(title="Billing", license=license))
```

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

## Testing

`MessagingTestCase` is a `neva.testing.TestCase` whose application *is* the
`App`, booted through its own lifespan — so `self.app`, the facade root and the
container a subscriber resolves from are one object, rather than the two a
hand-wired test ends up with.

```python
from neva.faststream.testing import MessagingTestCase


class TestDocuments(MessagingTestCase):
    broker: ClassVar[RabbitBroker]

    @override
    @classmethod
    def create_app(cls, config_path: Path) -> App:
        cls.broker = RabbitBroker()
        _ = cls.broker.subscriber("documents")(handle)
        app = App(cls.broker, config_path=config_path)
        _ = app.register(DocumentServiceProvider)
        return app

    async def test_it_processes(self) -> None:
        async with TestRabbitBroker(self.broker):
            await self.broker.publish({"id": 1}, "documents")

        assert Document.processed
```

Import it from `neva.faststream.testing`, never from `neva.faststream`: the
module pulls in pytest, which has no business in a production import.

Booting the application and connecting to a broker are separate things, and the
class does only the first — so it needs no live broker and imports no driver.
Start yours inside the test with `TestRabbitBroker`, or the equivalent for your
driver. Build the broker in `create_app` rather than at module level: one broker
belongs to one `App`, so a shared one fails in the second test class using it.

Register providers and declare subscribers in `create_app`, which runs before
anything boots. `App.register` returns `Err` once booted, and the brokers an
`App` owns are fixed at construction.

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

`MessagingTestCase` lives behind the `testing` extra:

```bash
uv add --dev "neva-faststream[testing]"
```

## 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
poe guidelines   # hold the agent guideline fragments to their contract
```

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

## Agent guidelines

This package contributes fragments to
[`neva-boost`](https://pypi.org/project/neva-boost/), which composes the Neva
ecosystem's agent guidelines into a project. `neva-boost install` writes them
alongside the core's, and `poe guidelines` holds this package's own to the
frontmatter contract.

Every fragment declares the version it needs, so one documenting an API your
installed version does not have is skipped rather than left to mislead an
agent.

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