feat(asyncapi): one channel per queue; lift the fastapi cap

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_<table>` LISTEN/NOTIFY channel)
+_Avoid_: topic, channel (reserve *channel* for the `outbox_<table>` 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

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

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

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

tests/test_unit.py

@@ -1731,18 +1731,45 @@ 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)
 
     @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: