From 58537bd1055ee0e1c1da517728d99b18d3dadd7f Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 20 Sep 2026 14:23:26 +0300 Subject: [PATCH 1/2] fix: adapt to faststream 0.7.6 and raise its floor 0.7.6 removed the try_it_out registry hook, made SubscriberSpec/PublisherSpec carry an address, made channel_labels abstract, narrowed dependencies to Sequence, and dropped add_task's return value. Registration is now declarative and the channel key comes from the base, so title_ names the channel too. --- CONTEXT.md | 5 +- faststream_redis_timers/__init__.py | 20 ------ faststream_redis_timers/broker.py | 2 +- faststream_redis_timers/publisher/usecase.py | 9 ++- faststream_redis_timers/registrator.py | 4 +- faststream_redis_timers/router.py | 4 +- faststream_redis_timers/subscriber/usecase.py | 12 +++- faststream_redis_timers/testing.py | 2 +- pyproject.toml | 3 +- tests/test_unit.py | 71 ++++++++++++++----- 10 files changed, 79 insertions(+), 53 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 0215987..9305fb0 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -24,8 +24,9 @@ The named channel a Timer is Scheduled on and that a subscriber consumes from. T derived per Topic; a Timer only ever exists on one. _Avoid_: queue — a Topic is not a work queue, and the difference matters: Timers on a Topic are ordered by Activation time, not arrival, and a Claimed Timer stays on the Topic rather than leaving -it. `channel` survives only where FastStream owns the spelling: the `raw_message` field and the -log-context key. In our own prose it is Topic. +it. `channel` survives only where FastStream owns the spelling: the `raw_message` field, the +log-context key, and `channel_labels` with the AsyncAPI document it feeds, where a Topic is a +channel — say *AsyncAPI channel* there. In our own prose it is Topic. **Activation time**: The single absolute UTC instant at which a Timer becomes Due. Exactly one of `activate_in` (a diff --git a/faststream_redis_timers/__init__.py b/faststream_redis_timers/__init__.py index 84cdd47..9b3d615 100644 --- a/faststream_redis_timers/__init__.py +++ b/faststream_redis_timers/__init__.py @@ -13,23 +13,3 @@ "TimersRoutePublisher", "TimersRouter", ] - -try: - import functools - import typing - - import faststream.asgi.factories.asyncapi.try_it_out - from faststream._internal.broker import BrokerUsecase - from faststream._internal.testing.broker import TestBroker - - 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]] - ]: - return {**original_get_broker_registry(), TimersBroker: TestTimersBroker} - - faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry = get_broker_registry # noqa: SLF001 -except Exception: # noqa: BLE001, S110 # pragma: no cover - pass diff --git a/faststream_redis_timers/broker.py b/faststream_redis_timers/broker.py index fa60d96..427a50d 100644 --- a/faststream_redis_timers/broker.py +++ b/faststream_redis_timers/broker.py @@ -97,7 +97,7 @@ def __init__( # noqa: PLR0913 start_timeout: float = 3.0, decoder: CustomCallable | None = None, parser: CustomCallable | None = None, - dependencies: Iterable[Dependant] = (), + dependencies: Sequence[Dependant] = (), middlewares: Sequence[type[BaseMiddleware] | BrokerMiddleware[TimerMessage]] = (), graceful_timeout: float | None = 15.0, routers: Sequence[Registrator[TimerMessage]] = (), diff --git a/faststream_redis_timers/publisher/usecase.py b/faststream_redis_timers/publisher/usecase.py index 1a4096a..fc5098f 100644 --- a/faststream_redis_timers/publisher/usecase.py +++ b/faststream_redis_timers/publisher/usecase.py @@ -22,13 +22,18 @@ class TimersPublisherSpecification(PublisherSpecification["TimersBrokerConfig", TimersPublisherSpecificationConfig]): # ty: ignore[unresolved-reference] @property - def name(self) -> str: + def full_topic(self) -> str: prefix = getattr(self._outer_config, "prefix", "") - return f"{prefix}{self.config.topic}:Publisher" + return f"{prefix}{self.config.topic}" + + @property + def name(self) -> str: + return f"{self.full_topic}:Publisher" def get_schema(self) -> dict[str, PublisherSpec]: return { self.name: PublisherSpec( + address=self.full_topic, description=self.config.description_, operation=Operation( message=Message( diff --git a/faststream_redis_timers/registrator.py b/faststream_redis_timers/registrator.py index c59805b..c705dc9 100644 --- a/faststream_redis_timers/registrator.py +++ b/faststream_redis_timers/registrator.py @@ -1,5 +1,5 @@ import warnings -from collections.abc import Iterable +from collections.abc import Sequence from typing import Any from fast_depends.dependencies import Dependant @@ -24,7 +24,7 @@ def subscriber( # ty: ignore[invalid-method-override] max_polling_interval: float = 5.0, max_concurrent: int = 5, lease_ttl: int = 30, - dependencies: Iterable[Dependant] = (), + dependencies: Sequence[Dependant] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, title_: str | None = None, diff --git a/faststream_redis_timers/router.py b/faststream_redis_timers/router.py index d514219..f87b889 100644 --- a/faststream_redis_timers/router.py +++ b/faststream_redis_timers/router.py @@ -45,7 +45,7 @@ def __init__( # noqa: PLR0913 max_concurrent: int = 5, lease_ttl: int = 30, publishers: Iterable[TimersRoutePublisher] = (), - dependencies: Iterable[Dependant] = (), + dependencies: Sequence[Dependant] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, title_: str | None = None, @@ -77,7 +77,7 @@ def __init__( # noqa: PLR0913 prefix: str = "", handlers: Iterable[TimersRoute] = (), *, - dependencies: Iterable[Dependant] = (), + dependencies: Sequence[Dependant] = (), middlewares: Sequence[BrokerMiddleware[TimerMessage]] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, diff --git a/faststream_redis_timers/subscriber/usecase.py b/faststream_redis_timers/subscriber/usecase.py index 936ca55..d7b9a03 100644 --- a/faststream_redis_timers/subscriber/usecase.py +++ b/faststream_redis_timers/subscriber/usecase.py @@ -29,13 +29,18 @@ class TimersSubscriberSpecification(SubscriberSpecification["TimersBrokerConfig", TimersSubscriberSpecificationConfig]): @property - def name(self) -> str: + def full_topic(self) -> str: prefix = getattr(self._outer_config, "prefix", "") - return f"{prefix}{self.config.topic}:{self.call_name}" + return f"{prefix}{self.config.topic}" + + @property + def channel_labels(self) -> list[str]: + return [self.full_topic] def get_schema(self) -> dict[str, SubscriberSpec]: return { self.name: SubscriberSpec( + address=self.full_topic, description=self.description, operation=Operation( message=Message( @@ -75,7 +80,8 @@ async def start(self) -> None: start_signal = anyio.Event() if self.calls: - consume_task = self.add_task(self._consume, (self._client,), {"start_signal": start_signal}) + self.add_task(self._consume, (self._client,), {"start_signal": start_signal}) + consume_task = self.tasks[-1] try: with anyio.fail_after(self._outer_config.start_timeout): await start_signal.wait() diff --git a/faststream_redis_timers/testing.py b/faststream_redis_timers/testing.py index 8160ef6..5c3008b 100644 --- a/faststream_redis_timers/testing.py +++ b/faststream_redis_timers/testing.py @@ -31,7 +31,7 @@ class ScheduledTimer: headers: dict[str, typing.Any] | None = None -class TestTimersBroker(TestBroker[TimersBroker, TimersBroker]): +class TestTimersBroker(TestBroker[TimersBroker, TimersBroker], broker=TimersBroker): scheduled_timers: list[ScheduledTimer] def __init__(self, broker: TimersBroker, **kwargs: typing.Any) -> None: diff --git a/pyproject.toml b/pyproject.toml index 2f733b8..050d16b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,8 +21,7 @@ classifiers = [ "Topic :: Software Development :: Libraries", ] dependencies = [ - # 0.7.6 makes SubscriberUsecase.add_task return None and drops _get_broker_registry; lift once this adapts. - "faststream>=0.7.1,<0.7.6", + "faststream>=0.7.6,<0.8", "redis>=5.0", "typing-extensions>=4.12.0", ] diff --git a/tests/test_unit.py b/tests/test_unit.py index a784a9e..56eca7a 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -6,9 +6,9 @@ from unittest.mock import AsyncMock, MagicMock, patch import anyio -import faststream.asgi.factories.asyncapi.try_it_out import pytest from faststream._internal.parser import DefaultCodec +from faststream._internal.testing.broker import find_test_broker from faststream.exceptions import IncorrectState from redis.asyncio.cluster import RedisCluster from redis.exceptions import NoScriptError @@ -31,9 +31,15 @@ # --- AsyncAPI try_it_out registry --- -def test_timers_broker_registered_in_try_it_out_registry() -> None: - registry = faststream.asgi.factories.asyncapi.try_it_out._get_broker_registry() # noqa: SLF001 - assert registry[TimersBroker] is TestTimersBroker +def test_timers_broker_registered_in_test_broker_registry() -> None: + """INVARIANT: a `TimersBroker` resolves to `TestTimersBroker` through FastStream's own registry. + + The AsyncAPI "try it out" page looks the test broker up by broker class, so a `TimersBroker` + that is absent from the registry silently loses that page. Registration is declarative — the + `broker=` argument on the class statement — so deleting it breaks this with nothing else to + notice, and a registry populated by hand goes stale against whatever FastStream does next. + """ + assert find_test_broker(TimersBroker()) is TestTimersBroker # --- ConnectionState --- @@ -177,27 +183,56 @@ async def test_publisher_request_raises() -> None: # --- TimersSubscriberSpecification.name / get_schema --- -def test_subscriber_specification_name_and_schema() -> None: +async def test_subscriber_specification_addresses_the_channel_by_its_full_topic() -> None: + """INVARIANT: the channel a subscriber emits is addressed by the topic it actually polls. + + The channel key carries the handler name with it, so it names nothing subscribable on its own; + `address` is the only field a reader can act on. Dropping the prefix breaks it just as surely + as omitting the field: a broker-prefixed deployment would publish an address no timer lands on. + """ + broker = TimersBroker() + router = TimersRouter(prefix="app:") + sub = router.subscriber("my-topic") + + @sub + async def handle(body: str) -> None: ... + + broker.include_router(router) + async with TestTimersBroker(broker): + schema = sub.specification.get_schema() + + assert {key: spec.address for key, spec in schema.items()} == {"app:my-topic:Handle": "app:my-topic"} + + +async def test_subscriber_title_names_the_channel_not_only_the_operation() -> None: + """`title_` names the channel, matching the publisher and every built-in broker.""" broker = TimersBroker() - sub = broker.subscriber("my-topic") - spec = sub.specification - name = spec.name - assert "my-topic" in name - schema = spec.get_schema() - assert schema # non-empty dict + sub = broker.subscriber("my-topic", title_="TimerIngest") + + @sub + async def handle(body: str) -> None: ... + + async with TestTimersBroker(broker): + assert list(sub.specification.get_schema()) == ["TimerIngest"] # --- TimersPublisherSpecification.name / get_schema --- -def test_publisher_specification_name_and_schema() -> None: +def test_publisher_specification_addresses_the_channel_by_its_full_topic() -> None: + """INVARIANT: the channel a publisher emits is addressed by the topic it actually writes to. + + Same reason as the subscriber: the key is `topic:Publisher`, which no client can publish to, + and an address that drops the broker prefix points at a topic nothing reads. + """ broker = TimersBroker() - pub = broker.publisher("my-topic") - spec = pub.specification - name = spec.name - assert "my-topic" in name - schema = spec.get_schema() - assert schema # non-empty dict + router = TimersRouter(prefix="app:") + pub = router.publisher("my-topic") + + broker.include_router(router) + schema = pub.specification.get_schema() + + assert {key: spec.address for key, spec in schema.items()} == {"app:my-topic:Publisher": "app:my-topic"} # --- TimerStreamMessage.nack --- From 7435bfb5a0705d54a9c5d363def04a207d2ad644 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 20 Sep 2026 14:32:56 +0300 Subject: [PATCH 2/2] fix(asyncapi): give the broker spec a non-empty server url Upstream collects channels only for brokers that reach its broker_servers mapping, which it fills inside `for url in specification.url`. TimersBroker passed url=[], so the whole document rendered blank while every per-endpoint get_schema() stayed correct. The url is rebuilt from the pool's connection parameters, never the caller's DSN, so no credential reaches the document. --- faststream_redis_timers/broker.py | 26 ++++++++++++++- tests/test_unit.py | 54 +++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/faststream_redis_timers/broker.py b/faststream_redis_timers/broker.py index 427a50d..91a35de 100644 --- a/faststream_redis_timers/broker.py +++ b/faststream_redis_timers/broker.py @@ -19,6 +19,7 @@ from faststream.specification.schema import BrokerSpec from faststream.specification.schema.extra import Tag, TagDict from redis.asyncio.cluster import RedisCluster +from redis.asyncio.connection import SSLConnection from typing_extensions import override from faststream_redis_timers.configs import ConnectionState, RedisClient, TimersBrokerConfig @@ -41,6 +42,29 @@ def _require_topic(topic: str) -> None: raise ValueError(msg) +def _spec_url(client: "RedisClient | None", timeline_key: str) -> list[str]: + """AsyncAPI server URL(s) for the broker spec. + + **Must be non-empty.** Upstream collects channels only for brokers that reach its + ``broker_servers`` mapping, which it fills inside ``for url in specification.url``; + an empty list renders a document with no servers, channels or operations while every + per-endpoint ``get_schema()`` stays correct. Rebuilt from the pool's connection + parameters rather than the caller's DSN, so no credential can reach the document. + """ + if client is None: + return [f"redis://timers/{timeline_key}"] + + connection_kwargs = client.connection_pool.connection_kwargs + if socket_path := connection_kwargs.get("path"): + return [f"unix://{socket_path}"] + + scheme = "rediss" if issubclass(client.connection_pool.connection_class, SSLConnection) else "redis" + host = connection_kwargs.get("host", "localhost") + port = connection_kwargs.get("port", 6379) + db = connection_kwargs.get("db", 0) + return [f"{scheme}://{host}:{port}/{db}"] + + class TimersParamsStorage(DefaultLoggerStorage): __max_msg_id_ln = -1 _max_channel_name = 7 @@ -142,7 +166,7 @@ def __init__( # noqa: PLR0913 ), ) specification = BrokerSpec( - url=[], + url=_spec_url(client, timeline_key), protocol="redis", protocol_version="5.0", description=description, diff --git a/tests/test_unit.py b/tests/test_unit.py index 56eca7a..619171c 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -10,6 +10,8 @@ from faststream._internal.parser import DefaultCodec from faststream._internal.testing.broker import find_test_broker from faststream.exceptions import IncorrectState +from faststream.specification import AsyncAPI +from redis.asyncio import Redis from redis.asyncio.cluster import RedisCluster from redis.exceptions import NoScriptError @@ -180,6 +182,58 @@ async def test_publisher_request_raises() -> None: await pub.request("x") +# --- AsyncAPI document assembly --- + + +async def test_asyncapi_document_carries_the_declared_channels() -> None: + """INVARIANT: a broker's endpoints reach the assembled AsyncAPI document, not just `get_schema()`. + + Upstream collects channels only for brokers that reach `broker_servers`, and it fills that + mapping inside `for url in specification.url`. A broker whose spec carries an empty `url` is + therefore skipped outright and renders a structurally blank document — no servers, channels or + operations — while every per-endpoint `get_schema()` stays correct, so specification-level tests + cannot see it. Emptying `BrokerSpec.url` breaks this again, silently. + """ + broker = TimersBroker(Redis.from_url("redis://cache.example:6399/3")) + sub = broker.subscriber("reminders") + + @sub + async def handle(body: str) -> None: ... + + broker.publisher("reminders") + + async with TestTimersBroker(broker): + spec = AsyncAPI(broker).to_specification().to_jsonable() + + assert sorted(spec["channels"]) == ["reminders:Handle", "reminders:Publisher"] + assert [server["host"] for server in spec["servers"].values()] == ["cache.example:6399"] + + +def test_specification_url_carries_no_credentials() -> None: + """INVARIANT: the AsyncAPI server URL is rebuilt from connection parameters, never the DSN. + + The document is published — FastStream serves it from the ASGI app — so a URL echoed back from + `Redis.from_url("redis://user:secret@...")` would publish the password. Reading the pool's + `username`/`password` into the URL, or passing the caller's DSN through, breaks it. + """ + broker = TimersBroker(Redis.from_url("redis://user:secret@cache.example:6399/3")) + + assert broker.specification.url == ["redis://cache.example:6399/3"] + + +@pytest.mark.parametrize( + ("client", "expected"), + [ + (None, "redis://timers/tl"), + (Redis.from_url("unix:///tmp/redis.sock?db=2"), "unix:///tmp/redis.sock"), + (Redis.from_url("rediss://cache.example:6379/0"), "rediss://cache.example:6379/0"), + ], + ids=["no-client", "unix-socket", "tls"], +) +def test_specification_url_shapes(client: Redis | None, expected: str) -> None: + assert TimersBroker(client, timeline_key="tl").specification.url == [expected] + + # --- TimersSubscriberSpecification.name / get_schema ---