From 48cc4641f43dec989ae2b06909f79edecf361446 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 20 Sep 2026 13:30:32 +0300 Subject: [PATCH] feat(asyncapi): one channel per queue; lift the fastapi cap Two independent changes that both fell out of the faststream 0.7.6 bump. AsyncAPI channels, closing #181. 0.7.6 reworked the specification model around `channel_labels` (one label per channel) and a required `SubscriberSpec.address`. The subscriber kept filing every queue under a single channel keyed by the joined queue list, which forced a synthetic address: `"orders,shipments"` names nothing a consumer can subscribe to. Dropping the `name` override in favour of the base's `_channel_key` also fixes `title_`, which is public and documented but named only the operation, unlike the publisher and every built-in broker. The fastapi cap. It existed because fastapi 0.140 made `Dependant` a slotted dataclass and broke faststream's integration (ag2ai/faststream#2959). That issue closed completed on 2026-07-28 and the fix shipped no later than faststream 0.7.4, which is below the 0.7.6 floor this package already requires, so the pairing can no longer regress within the supported range. Verified against fastapi 0.141.1. `CONTEXT.md` reserves *channel* for the LISTEN/NOTIFY channel, so it now records the one place upstream's vocabulary overrides ours. --- CONTEXT.md | 4 ++- docs/usage/subscriber.md | 8 ++++++ faststream_outbox/subscriber/usecase.py | 28 +++++++++++-------- pyproject.toml | 8 ++---- tests/test_unit.py | 37 +++++++++++++++++++++---- 5 files changed, 62 insertions(+), 23 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index e03373e..4fdbb54 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -18,7 +18,9 @@ _Avoid_: job, task, event **Queue**: The `queue` column value a subscriber filters on. Not a separate object; there is nothing to declare or create. -_Avoid_: topic, channel (reserve *channel* for the `outbox_` LISTEN/NOTIFY channel) +_Avoid_: topic, channel (reserve *channel* for the `outbox_
` LISTEN/NOTIFY channel). +One exception, upstream's and not ours: `channel_labels` and the AsyncAPI document it feeds +call a queue a channel. Say *AsyncAPI channel* there. **Lease**: A time-bounded claim on a row, held as the `(acquired_token, acquired_at)` pair. It expires on its diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index 24942e4..5071222 100644 --- a/docs/usage/subscriber.md +++ b/docs/usage/subscriber.md @@ -31,6 +31,10 @@ The subscriber claims rows from any of its queues in a single fetch. Its [connection budget](#connection-budget) is unchanged — `max_workers + 1` pool connections regardless of how many queues it serves. +In the AsyncAPI document it appears as one channel per queue +(`orders:Handle`, `refunds:Handle`), each addressed by that queue, rather +than one channel for the subscriber. + Do **not** register two subscribers on the **same** queue: they compete for the same rows, and registration emits a warning to that effect. To run more than one handler over a queue, attach them to a single subscriber; to scale @@ -115,6 +119,10 @@ The table above lists the outbox-specific knobs. The standard FastStream subscriber kwargs pass through unchanged too: `dependencies`, `parser`, `decoder`, and the AsyncAPI `title_` / `description_` / `include_in_schema`. +`title_` names both the AsyncAPI channel and its operation. On a subscriber +spanning several queues it prefixes each channel (`Ingest:orders`, +`Ingest:refunds`), since one title cannot name several channels on its own. + ## Slow handlers — dedicated queue When a handler's tail latency exceeds the subscriber's `lease_ttl_seconds`, diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 752ced4..cdee831 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -161,28 +161,34 @@ def _render_last_exception( class OutboxSubscriberSpecification(SubscriberSpecification["OutboxBrokerConfig", OutboxSubscriberSpecificationConfig]): @property def channel_labels(self) -> list[str]: - """One channel, keyed by every queue this subscriber drains.""" - return [",".join(self.config.queues)] + """The queues this subscriber drains, one per AsyncAPI channel. - @property - def name(self) -> str: - return f"{self.channel_labels[0]}:{self.call_name}" + Deduped through a dict rather than a set: set order varies per process and would + reach the document as the order of its channels. + """ + return list(dict.fromkeys(self.config.queues)) def get_schema(self) -> dict[str, SubscriberSpec]: - return { - self.name: SubscriberSpec( - address=self.channel_labels[0], + payloads = self.get_payloads() + labels = self.channel_labels + split = len(labels) > 1 + + schema = {} + for queue in labels: + key = self._channel_key(queue, split=split) + schema[key] = SubscriberSpec( + address=queue, description=self.description, operation=Operation( message=Message( - title=f"{self.name}:Message", - payload=resolve_payloads(self.get_payloads()), + title=f"{key}:Message", + payload=resolve_payloads(payloads), ), bindings=None, ), bindings=None, ) - } + return schema class OutboxSubscriber(TasksMixin, SubscriberUsecase[OutboxInnerMessage]): diff --git a/pyproject.toml b/pyproject.toml index 219e9bd..13606d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,11 +29,7 @@ dependencies = [ [project.optional-dependencies] asyncpg = ["asyncpg>=0.29"] validate = ["alembic>=1.13"] -# Upper cap: fastapi 0.140 made Dependant a slotted dataclass, which breaks -# faststream's FastAPI integration (ag2ai/faststream#2959). faststream declares -# no fastapi dependency of its own, so this extra is the only place the pairing -# can be constrained. Lift the cap and raise the faststream floor once fixed. -fastapi = ["fastapi>=0.95,<0.140"] +fastapi = ["fastapi>=0.95"] prometheus = ["prometheus-client>=0.19"] opentelemetry = ["opentelemetry-api>=1.20", "opentelemetry-sdk>=1.20"] all = ["faststream-outbox[asyncpg,validate,fastapi,prometheus,opentelemetry]"] @@ -45,7 +41,7 @@ dev = [ "pytest-cov", "asyncpg>=0.29", "alembic>=1.13", - "fastapi>=0.95,<0.140", # see the cap note on the fastapi extra + "fastapi>=0.95", "faststream[kafka]>=0.7.6,<0.8", "httpx2>=2.2", "prometheus-client>=0.19", diff --git a/tests/test_unit.py b/tests/test_unit.py index 81633e6..3af2ac6 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -1731,7 +1731,14 @@ async def handler(body: str) -> None: ... assert route is not None -def test_subscriber_specification_name_lists_queues() -> None: +async def test_subscriber_specification_emits_one_channel_per_queue() -> None: + """INVARIANT: a multi-queue subscriber emits one channel per queue, each addressed by that queue. + + Folding them back into a single channel keyed by the joined queue list forces a synthetic + address: ``"orders,shipments"`` names nothing a consumer can subscribe to, and AsyncAPI offers + no way to read one address as several. The joined form predates ``SubscriberSpec.address``, + when the channel key was the only queue information the document carried. + """ metadata = MetaData() t = make_outbox_table(metadata) broker = OutboxBroker(outbox_table=t) @@ -1739,10 +1746,30 @@ def test_subscriber_specification_name_lists_queues() -> None: @broker.subscriber(["orders", "shipments"]) async def handle(body: str) -> None: ... - sub = next(iter(broker._subscribers)) # noqa: SLF001 - name = sub.specification.name - assert "orders" in name - assert "shipments" in name + async with TestOutboxBroker(broker): + sub = next(iter(broker._subscribers)) # noqa: SLF001 + schema = sub.specification.get_schema() + + assert {key: spec.address for key, spec in schema.items()} == { + "orders:Handle": "orders", + "shipments:Handle": "shipments", + } + + +async def test_subscriber_title_names_the_channel_not_only_the_operation() -> None: + """``title_`` names the channel as well, matching the publisher and every built-in broker.""" + metadata = MetaData() + t = make_outbox_table(metadata) + broker = OutboxBroker(outbox_table=t) + + @broker.subscriber("orders", title_="OrderIngest") + async def handle(body: str) -> None: ... + + async with TestOutboxBroker(broker): + spec = AsyncAPI(broker).to_specification().to_jsonable() + + assert list(spec["channels"]) == ["OrderIngest"] + assert spec["channels"]["OrderIngest"]["address"] == "orders" async def test_subscriber_specification_get_schema() -> None: