From eed7b5da78e1655e8726fc5bcfc46d5b70737540 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 20 Sep 2026 13:20:14 +0300 Subject: [PATCH] chore(deps): adapt to faststream 0.7.6 and raise its floor faststream 0.7.6 replaced the private ASGI try-it-out registry with a `TestBroker.__init_subclass__(broker=...)` hook, made `SubscriberSpec` / `PublisherSpec` carry a required `address`, made `channel_labels` abstract, narrowed `dependencies` to `Sequence`, and turned `__aiter__` back into a plain `def`. The registry monkeypatch goes away entirely in favour of the declarative hook, which also drops a reach into a private symbol. `TestBroker` did not accept the `broker=` kwarg before 0.7.6, so `import faststream_outbox` now fails outright on 0.7.5; the floor moves from 0.7.1 to 0.7.6 to say so. Nothing in CI resolves at the lower bound, which is why the previous floor went stale unnoticed. `publisher.mock` now raises `SetupError` outside a test broker rather than answering with a reset mock, which exposed a vacuous assertion in the relay dual-fire test: it checked `assert_not_called()` after leaving the `TestKafkaBroker` context, where the mock had already been cleared. Moved inside, it tests something. The subscriber keeps its single comma-joined AsyncAPI channel rather than splitting one per queue the way the built-in brokers now do. That is a document-shape decision, not a dependency one: #181. --- faststream_outbox/__init__.py | 29 -------------------- faststream_outbox/broker.py | 2 +- faststream_outbox/publisher/specification.py | 1 + faststream_outbox/registrator.py | 4 +-- faststream_outbox/router.py | 4 +-- faststream_outbox/subscriber/usecase.py | 14 ++++++---- faststream_outbox/testing.py | 6 ++-- pyproject.toml | 4 +-- tests/test_integration.py | 6 ++-- tests/test_unit.py | 15 +++++----- 10 files changed, 31 insertions(+), 54 deletions(-) diff --git a/faststream_outbox/__init__.py b/faststream_outbox/__init__.py index 838f8cf..1e8615d 100644 --- a/faststream_outbox/__init__.py +++ b/faststream_outbox/__init__.py @@ -1,9 +1,3 @@ -import functools -import typing - -from faststream._internal.broker import BrokerUsecase -from faststream._internal.testing.broker import TestBroker - from faststream_outbox.autovacuum import outbox_autovacuum_ddl from faststream_outbox.broker import OutboxBroker from faststream_outbox.message import OutboxMessage @@ -39,26 +33,3 @@ "make_outbox_table", "outbox_autovacuum_ddl", ] - -try: - # S4: import inside the guard too — if upstream moves/removes the module, this - # raises ImportError here and is tolerated, instead of breaking ``import - # faststream_outbox`` from an unguarded top-level import. - import faststream.asgi.factories.asyncapi.try_it_out - - original_get_broker_registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry # noqa: SLF001 - - @functools.lru_cache(maxsize=1) - def get_broker_registry() -> dict[ - type[BrokerUsecase[typing.Any, typing.Any]], - type[TestBroker[typing.Any, typing.Any]], - ]: - # BrokerUsecase is invariant on its config type, so OutboxBrokerConfig won't unify with BrokerConfig. - return {**original_get_broker_registry(), OutboxBroker: TestOutboxBroker} # ty: ignore[invalid-return-type] - - faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry = get_broker_registry # noqa: SLF001 -except (AttributeError, ImportError): # pragma: no cover - # FastStream's private ASGI try-it-out registry is best-effort wiring; - # tolerate breakage if upstream renames/moves the symbol but surface other - # errors (config, type) loudly so we notice them in CI. - pass diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index d2e76e3..dadf969 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -134,7 +134,7 @@ def __init__( # noqa: PLR0913 dlq_table: "Table | None" = None, decoder: CustomCallable | None = None, parser: CustomCallable | None = None, - dependencies: Iterable["Dependant"] = (), + dependencies: Sequence["Dependant"] = (), middlewares: Sequence[type[BaseMiddleware] | BrokerMiddleware[OutboxInnerMessage]] = (), graceful_timeout: float | None = 15.0, routers: Sequence[Registrator[OutboxInnerMessage]] = (), diff --git a/faststream_outbox/publisher/specification.py b/faststream_outbox/publisher/specification.py index 1b4e78b..1146425 100644 --- a/faststream_outbox/publisher/specification.py +++ b/faststream_outbox/publisher/specification.py @@ -21,6 +21,7 @@ def get_schema(self) -> dict[str, PublisherSpec]: payloads = self.get_payloads() return { self.name: PublisherSpec( + address=self.config.queue, description=self.config.description_, operation=Operation( message=Message( diff --git a/faststream_outbox/registrator.py b/faststream_outbox/registrator.py index 86bf57b..ccac147 100644 --- a/faststream_outbox/registrator.py +++ b/faststream_outbox/registrator.py @@ -1,5 +1,5 @@ import warnings -from collections.abc import Iterable +from collections.abc import Sequence from typing import TYPE_CHECKING, Any from faststream._internal.broker.registrator import Registrator @@ -54,7 +54,7 @@ def subscriber( # ty: ignore[invalid-method-override] terminal_flush_batch_size: int = 1, ack_policy: AckPolicy | None = None, propagate_inbound_headers: bool = False, - dependencies: Iterable["Dependant"] = (), + dependencies: Sequence["Dependant"] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, title_: str | None = None, diff --git a/faststream_outbox/router.py b/faststream_outbox/router.py index 379d3f2..a919e1a 100644 --- a/faststream_outbox/router.py +++ b/faststream_outbox/router.py @@ -35,7 +35,7 @@ def __init__( # noqa: PLR0913 terminal_flush_batch_size: int = 1, ack_policy: AckPolicy | None = None, propagate_inbound_headers: bool = False, - dependencies: Iterable["Dependant"] = (), + dependencies: Sequence["Dependant"] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, title_: str | None = None, @@ -77,7 +77,7 @@ def __init__( self, handlers: Iterable[OutboxRoute] = (), *, - dependencies: Iterable["Dependant"] = (), + dependencies: Sequence["Dependant"] = (), middlewares: Sequence[BrokerMiddleware[OutboxInnerMessage]] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 7b04876..752ced4 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -159,14 +159,19 @@ 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)] + @property def name(self) -> str: - joined = ",".join(self.config.queues) - return f"{joined}:{self.call_name}" + return f"{self.channel_labels[0]}:{self.call_name}" def get_schema(self) -> dict[str, SubscriberSpec]: return { self.name: SubscriberSpec( + address=self.channel_labels[0], description=self.description, operation=Operation( message=Message( @@ -923,13 +928,12 @@ async def get_one(self, *, timeout: float = 5.0) -> typing.NoReturn: raise NotImplementedError(_UNSUPPORTED_PEEK_MSG) @override - async def __aiter__(self) -> AsyncIterator["StreamMessage[OutboxInnerMessage]"]: + def __aiter__(self) -> AsyncIterator["StreamMessage[OutboxInnerMessage]"]: # Native FakeStream subscribers (e.g. redis ListSubscriber.__aiter__) implement # this against a blocking pop; for the outbox, a true peek would acquire a lease # and bump deliveries_count — surprising semantics for a "look but don't touch" # API. Route operators at ``broker.fetch_unprocessed`` instead, which is - # lease-free and doesn't mutate row state. Matches the base's no-yield shape so - # the override stays a coroutine returning AsyncIterator (not an async generator). + # lease-free and doesn't mutate row state. raise NotImplementedError(_UNSUPPORTED_PEEK_MSG) @override diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 5e92164..cb47424 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -568,8 +568,7 @@ async def fake_publish_batch( return fake_publish_batch -# BrokerUsecase is invariant on its config type, so OutboxBrokerConfig won't unify with BrokerConfig. -class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker]): # ty: ignore[invalid-type-arguments] +class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker], broker=OutboxBroker): """Test harness for ``OutboxBroker``. Two dispatch modes. Default (``run_loops=False``): ``broker.publish`` synchronously drives the matching @@ -682,8 +681,7 @@ def _fake_start(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing. # Skip the parent's publisher iteration — see ``create_publisher_fake_subscriber`` # for why. We still need to fan out ``_post_start`` on subscribers so their # call models build (matches what TestBroker._fake_start does last). - # BrokerUsecase is invariant on its config type; this only iterates broker.subscribers. - patch_broker_calls(broker) # ty: ignore[invalid-argument-type] + patch_broker_calls(broker) for subscriber in broker.subscribers: subscriber._post_start() # noqa: SLF001 broker._warn_on_unstarted_foreign_publishers() # noqa: SLF001 diff --git a/pyproject.toml b/pyproject.toml index 4d4f73e..219e9bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,7 @@ classifiers = [ "Topic :: Software Development :: Libraries", ] dependencies = [ - "faststream>=0.7.1,<0.8", + "faststream>=0.7.6,<0.8", "sqlalchemy[asyncio]>=2.0", "typing-extensions>=4.12.0", ] @@ -46,7 +46,7 @@ dev = [ "asyncpg>=0.29", "alembic>=1.13", "fastapi>=0.95,<0.140", # see the cap note on the fastapi extra - "faststream[kafka]>=0.7.1,<0.8", + "faststream[kafka]>=0.7.6,<0.8", "httpx2>=2.2", "prometheus-client>=0.19", "opentelemetry-api>=1.20", diff --git a/tests/test_integration.py b/tests/test_integration.py index f1a5e1f..4c4a1dc 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1947,8 +1947,10 @@ def spy_log(*args: Any, **kwargs: Any) -> Any: await broker_outbox.publish({"x": 1}, queue="relay_queue", session=session) await _wait_until(lambda: bool(errors), timeout=10.0) - # Guard fired before the chain → the foreign Kafka publish never happened. - publisher_kafka.mock.assert_not_called() + # Guard fired before the chain → the foreign Kafka publish never happened. + # Asserted inside the TestKafkaBroker context: outside it, `.mock` raises SetupError. + publisher_kafka.mock.assert_not_called() + # Row left in place (lease held, not deleted) for lease-expiry retry. assert await _row_count(pg_engine, outbox_table) == 1, "config-error row must be left for lease-expiry, not deleted" # And the _OutboxConfigError was logged at ERROR by the worker loop. diff --git a/tests/test_unit.py b/tests/test_unit.py index aa3f719..81633e6 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -11,10 +11,10 @@ from contextlib import asynccontextmanager from unittest.mock import AsyncMock, MagicMock, patch -import faststream.asgi.factories.asyncapi.try_it_out import pytest from faststream._internal.parser import DefaultCodec from faststream._internal.producer import ProducerProto +from faststream._internal.testing.broker import find_test_broker from faststream.exceptions import IncorrectState from faststream.middlewares import AckPolicy from faststream.response.publish_type import PublishType @@ -72,9 +72,10 @@ from faststream_outbox.testing import FakeOutboxClient, FakeOutboxProducer -def test_outbox_broker_registered_in_try_it_out_registry() -> None: - registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry() # noqa: SLF001 - assert registry[OutboxBroker] is TestOutboxBroker # ty: ignore[invalid-argument-type] +def test_outbox_broker_registered_in_test_broker_registry() -> None: + metadata = MetaData() + broker = OutboxBroker(outbox_table=make_outbox_table(metadata)) + assert find_test_broker(broker) is TestOutboxBroker def _make_broker(engine: object | None = None, table_name: str = "outbox") -> OutboxBroker: @@ -1831,7 +1832,7 @@ async def handle(body: str) -> None: ... # __aiter__ is also unsupported (was silently abstract-inherited before B6). with pytest.raises(NotImplementedError, match="fetch_unprocessed"): - await sub.__aiter__() + sub.__aiter__() # _make_response_publisher returns an OutboxFakePublisher wired to the producer # so handlers can ``return OutboxResponse(...)``. @@ -4179,7 +4180,7 @@ async def handle(body: dict) -> None: ... # registered for the spec; never invo broker.publisher("events") - spec = AsyncAPI(broker).to_specification().to_jsonable() # ty: ignore[invalid-argument-type] # BrokerUsecase invariance + spec = AsyncAPI(broker).to_specification().to_jsonable() assert spec["servers"], "AsyncAPI servers must not be empty (url=[] regression)" channel_keys = " ".join(spec["channels"]) assert "orders" in channel_keys, f"subscriber channel missing: {list(spec['channels'])}" @@ -4198,7 +4199,7 @@ async def handle(body: dict) -> None: ... # registered for the spec; never invo broker.publisher("hidden_events", include_in_schema=False) - spec = AsyncAPI(broker).to_specification().to_jsonable() # ty: ignore[invalid-argument-type] # BrokerUsecase invariance + spec = AsyncAPI(broker).to_specification().to_jsonable() channel_keys = " ".join(spec["channels"]) assert "orders" in channel_keys # the included subscriber is present… assert "hidden_events" not in channel_keys # …the excluded publisher is not