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

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

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.

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.

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:

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:

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

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:

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:

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

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.

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

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:

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

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