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

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.