neva-faststream

FastStream integration for the Neva framework — the messaging counterpart to neva-fastapi.

It marries FastStream 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

# 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")

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:

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 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:

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.

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.

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.

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:

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:

from neva.faststream import inject

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

Install

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

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 with gitmoji prefixes, enforced by cz_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:

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.