From 3633f488f984b64f378d87da8a3c45dd923d551a Mon Sep 17 00:00:00 2001 From: Tom Kimber Date: Wed, 7 Oct 2026 22:33:24 +0100 Subject: [PATCH] feat(asyncfast): add request/reply asyncapi support --- .../asyncfast/docs/examples/request_reply.py | 18 + .../examples/request_reply_dynamic_address.py | 24 ++ packages/asyncfast/docs/index.rst | 1 + packages/asyncfast/docs/request_reply.rst | 86 ++++ packages/asyncfast/src/asyncfast/__init__.py | 6 + packages/asyncfast/src/asyncfast/_asyncapi.py | 76 +++- .../asyncfast/src/asyncfast/_asyncfast.py | 25 +- packages/asyncfast/src/asyncfast/_channel.py | 104 ++++- .../tests_asyncfast/test_asyncapi.py | 370 ++++++++++++++++++ .../asyncfast/tests_asyncfast/test_errors.py | 108 +++++ 10 files changed, 804 insertions(+), 14 deletions(-) create mode 100644 packages/asyncfast/docs/examples/request_reply.py create mode 100644 packages/asyncfast/docs/examples/request_reply_dynamic_address.py create mode 100644 packages/asyncfast/docs/request_reply.rst diff --git a/packages/asyncfast/docs/examples/request_reply.py b/packages/asyncfast/docs/examples/request_reply.py new file mode 100644 index 00000000..699c9d22 --- /dev/null +++ b/packages/asyncfast/docs/examples/request_reply.py @@ -0,0 +1,18 @@ +from collections.abc import AsyncGenerator +from dataclasses import dataclass + +from asyncfast import AsyncFast +from asyncfast import Message +from asyncfast import Reply + +app = AsyncFast() + + +@dataclass +class Pong(Message, address="pong"): + payload: str + + +@app.channel("ping", reply=Reply()) +async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") diff --git a/packages/asyncfast/docs/examples/request_reply_dynamic_address.py b/packages/asyncfast/docs/examples/request_reply_dynamic_address.py new file mode 100644 index 00000000..ad3e639a --- /dev/null +++ b/packages/asyncfast/docs/examples/request_reply_dynamic_address.py @@ -0,0 +1,24 @@ +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from typing import Annotated + +from asyncfast import AsyncFast +from asyncfast import Message +from asyncfast import Reply +from asyncfast import ReplyTo + +app = AsyncFast() + + +@dataclass +class Pong(Message, address="{reply_to}"): + reply_to: str + payload: str + + +@app.channel("ping", reply=Reply()) +async def ping( + payload: str, + reply_to: Annotated[str, ReplyTo(alias="replyTo")], +) -> AsyncGenerator[Pong, None]: + yield Pong(reply_to=reply_to, payload="pong") diff --git a/packages/asyncfast/docs/index.rst b/packages/asyncfast/docs/index.rst index 5f3a49a6..bdcbbf40 100644 --- a/packages/asyncfast/docs/index.rst +++ b/packages/asyncfast/docs/index.rst @@ -83,6 +83,7 @@ Taking ideas from: receiving sending + request_reply message_send_router dependencies lifespan diff --git a/packages/asyncfast/docs/request_reply.rst b/packages/asyncfast/docs/request_reply.rst new file mode 100644 index 00000000..5d2ca8cb --- /dev/null +++ b/packages/asyncfast/docs/request_reply.rst @@ -0,0 +1,86 @@ +############### + Request/Reply +############### + +Request/reply is where a requester sends a message and waits for a response, with the response sent back to the +requester. In AsyncFast the reply of the interaction is the message the channel handler sends, so the pattern is +declared by passing :py:class:`Reply` to the channel. + +************* + Basic Usage +************* + +Marking a channel with :py:class:`Reply` adds the AsyncAPI Operation Reply object to the receive operation, where the +reply channel is the channel of the message the handler sends: + +.. async-fast-example:: examples/request_reply.py + +The handler must send exactly one message type, otherwise an +:py:class:`InvalidChannelDefinitionError` is raised when the channel is registered, that is, when the ``@app.channel(...)`` decorator is applied. + +:py:class:`Reply` works with either sending style, so the :py:class:`MessageSender` can be used instead of yielding: + +.. code-block:: python + + @app.channel("ping", reply=Reply()) + async def ping(message_sender: MessageSender[Pong]) -> None: + await message_sender.send(Pong(payload="pong")) + +*********************** + Dynamic Reply Address +*********************** + +When the address of the reply isn't known at design time, it is determined at runtime from the request itself. This +is typically done by the requester including a ``replyTo`` header. Marking that header with :py:class:`ReplyTo` +tells AsyncFast it carries the reply address, so the AsyncAPI reply address is derived from it and the header only +needs to be written once: + +.. async-fast-example:: examples/request_reply_dynamic_address.py + +:py:class:`ReplyTo` behaves like :py:class:`Header`, so the value is injected into the handler, and it can equally be +used in a dependency of the channel. A channel can have at most one :py:class:`ReplyTo` header. + +The derived location is an AsyncAPI runtime expression, ``$message.header#/replyTo`` in the example above. To describe +the address, or to point at a location that isn't a header marked with :py:class:`ReplyTo`, pass a +:py:class:`ReplyAddress` to :py:class:`Reply`. The location must be either ``$message.header`` or +``$message.payload`` followed by a JSON Pointer. As a shortcut, a location can also be passed as a string: + +.. code-block:: python + + @app.channel("ping", reply=Reply(address="$message.payload#/replyTo")) + async def ping(payload: PingPayload) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") + +The reply address is documentation only, so the handler is still responsible for sending to the address the requester +supplied. Binding the :py:class:`ReplyTo` header to a parameter of the reply channel address, as in the example above, +is the most direct way of doing this. + +.. note:: + + AsyncAPI 3.0 requires that, when a reply address is specified, the address of the channel referenced by the reply + is ``null``. When :py:class:`Reply` is given an address, the reply channel is therefore rendered with a ``null`` + address, with the reply address runtime expression carrying the routing information. The reply message should not + be reused as the send message of a standalone channel, since its channel address is rendered as ``null``. + +******************** + Same Reply Address +******************** + +The request and the reply can also share an address, which is a reply channel using the same address as the request +channel: + +.. code-block:: python + + @dataclass + class Pong(Message, address="ping"): + payload: str + + + @app.channel("ping", reply=Reply()) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") + +.. note:: + + AsyncFast generates a channel per message sent, so the reply is a new channel which happens to reuse the request + address, rather than the request channel itself. diff --git a/packages/asyncfast/src/asyncfast/__init__.py b/packages/asyncfast/src/asyncfast/__init__.py index e18a89d9..52dfabbb 100644 --- a/packages/asyncfast/src/asyncfast/__init__.py +++ b/packages/asyncfast/src/asyncfast/__init__.py @@ -5,6 +5,9 @@ from asyncfast._channel import MessageSender from asyncfast._channel import Parameter from asyncfast._channel import Payload +from asyncfast._channel import Reply +from asyncfast._channel import ReplyAddress +from asyncfast._channel import ReplyTo from asyncfast._message import Message from asyncfast._utils import ChannelNotFoundError from asyncfast._utils import InvalidChannelDefinitionError @@ -18,6 +21,9 @@ "MessageSender", "Parameter", "Payload", + "Reply", + "ReplyAddress", + "ReplyTo", "Message", "ChannelNotFoundError", ] diff --git a/packages/asyncfast/src/asyncfast/_asyncapi.py b/packages/asyncfast/src/asyncfast/_asyncapi.py index 3afb18cd..07bb844a 100644 --- a/packages/asyncfast/src/asyncfast/_asyncapi.py +++ b/packages/asyncfast/src/asyncfast/_asyncapi.py @@ -20,8 +20,11 @@ from asyncfast._channel import MessageSenderResolver from asyncfast._channel import Payload from asyncfast._channel import PayloadResolver +from asyncfast._channel import Reply +from asyncfast._channel import ReplyAddress from asyncfast._channel import Resolver from asyncfast._message import Message +from asyncfast._utils import InvalidChannelDefinitionError from pydantic import BaseModel from pydantic import create_model from pydantic import TypeAdapter @@ -49,14 +52,15 @@ class MessageDefinition: headers: Sequence[tuple[str, type[Any], Any]] bindings: Sequence[tuple[str, str, type[Any], CoreSchema]] payload: tuple[type[Any], TypeAdapter[Any], Payload] | None + dynamic_address: bool = False @property def channel_definition(self) -> dict[str, Any]: definition = { - "address": self.address, + "address": None if self.dynamic_address else self.address, "messages": {self.name: {"$ref": f"#/components/messages/{self.name}"}}, } - if self.parameters: + if self.parameters and not self.dynamic_address: definition["parameters"] = {name: {} for name in self.parameters} return definition @@ -175,6 +179,64 @@ def payload(self) -> PayloadResolver[Any] | None: return payloads[0] return None + @cached_property + def reply(self) -> Reply | None: + return self.channel.reply + + @cached_property + def reply_address(self) -> ReplyAddress | None: + if self.reply is None: + return None + if isinstance(self.reply.address, ReplyAddress): + return self.reply.address + + reply_to_names = tuple( + dict.fromkeys( + resolver.name + for resolver in self.resolvers + if isinstance(resolver, HeaderResolver) and resolver.reply_to + ) + ) + if len(reply_to_names) > 1: + raise InvalidChannelDefinitionError( + "Channel must have no more than 1 reply to header" + ) + if not reply_to_names: + return None + pointer = reply_to_names[0].replace("~", "~0").replace("/", "~1") + return ReplyAddress(f"$message.header#/{pointer}") + + @cached_property + def reply_definition(self) -> dict[str, Any] | None: + if self.reply is None: + return None + + messages = tuple(dict.fromkeys(self.send_messages)) + if len(messages) != 1: + raise InvalidChannelDefinitionError( + "Channel with reply must send exactly one message type" + ) + (message,) = messages + + definition: dict[str, Any] = {} + + reply_address = self.reply_address + if isinstance(reply_address, ReplyAddress): + definition["address"] = { + "location": reply_address.location, + **( + {"description": reply_address.description} + if reply_address.description + else {} + ), + } + + definition["channel"] = {"$ref": f"#/channels/{message.__name__}"} + definition["messages"] = [ + {"$ref": f"#/channels/{message.__name__}/messages/{message.__name__}"} + ] + return definition + def generate_send_messages(self) -> Generator[type[Message], None, None]: signature = inspect.signature(self.channel.func) @@ -207,6 +269,7 @@ def send_messages(self) -> Sequence[type[Message]]: @cached_property def send_message_definitions(self) -> Sequence[MessageDefinition]: + dynamic_address = self.reply_address is not None return tuple( MessageDefinition( message.__name__, @@ -231,6 +294,7 @@ def send_message_definitions(self) -> Sequence[MessageDefinition]: if message.__payload__ else None ), + dynamic_address, ) for message in self.send_messages ) @@ -323,11 +387,17 @@ def generate_operations( channel_definitions: Iterable[ChannelDefinition], ) -> Generator[tuple[str, dict[str, Any]], None, None]: for channel_definition in channel_definitions: - yield f"receive{channel_definition.title}", { + receive_operation: dict[str, Any] = { "action": "receive", "channel": {"$ref": f"#/channels/{channel_definition.title}"}, } + reply_definition = channel_definition.reply_definition + if reply_definition is not None: + receive_operation["reply"] = reply_definition + + yield f"receive{channel_definition.title}", receive_operation + for message in channel_definition.send_messages: yield f"send{message.__name__}", { "action": "send", diff --git a/packages/asyncfast/src/asyncfast/_asyncfast.py b/packages/asyncfast/src/asyncfast/_asyncfast.py index 2b3aa2c7..2502a9b8 100644 --- a/packages/asyncfast/src/asyncfast/_asyncfast.py +++ b/packages/asyncfast/src/asyncfast/_asyncfast.py @@ -14,8 +14,12 @@ from amgi_types import LifespanShutdownCompleteEvent from amgi_types import LifespanStartupCompleteEvent from amgi_types import Scope +from asyncfast._asyncapi import ChannelDefinition from asyncfast._asyncapi import get_asyncapi from asyncfast._channel import ChannelRouter +from asyncfast._channel import get_channel +from asyncfast._channel import Reply +from asyncfast._utils import InvalidChannelDefinitionError from asyncfast.middleware.errors import ServerErrorMiddleware P = ParamSpec("P") @@ -67,13 +71,26 @@ def title(self) -> str: def version(self) -> str: return self._version - def channel(self, address: str) -> Callable[[DecoratedCallable], DecoratedCallable]: - return partial(self._add_channel, address) + def channel( + self, + address: str, + *, + reply: Reply | None = None, + ) -> Callable[[DecoratedCallable], DecoratedCallable]: + return partial(self._add_channel, address, reply) def _add_channel( - self, address: str, function: DecoratedCallable + self, address: str, reply: Reply | None, function: DecoratedCallable ) -> DecoratedCallable: - self._router.add_channel(address, function) + channel = get_channel(function, address, reply) + if ( + reply is not None + and len(set(ChannelDefinition(channel).send_messages)) != 1 + ): + raise InvalidChannelDefinitionError( + "Channel with reply must send exactly one message type" + ) + self._router.add_route(address, channel) return function async def __call__( diff --git a/packages/asyncfast/src/asyncfast/_channel.py b/packages/asyncfast/src/asyncfast/_channel.py index 102b55f3..63a9ced9 100644 --- a/packages/asyncfast/src/asyncfast/_channel.py +++ b/packages/asyncfast/src/asyncfast/_channel.py @@ -42,6 +42,9 @@ _FIELD_PATTERN = re.compile(r"^[A-Za-z0-9_\-]+$") _PARAMETER_PATTERN = re.compile(r"{(.*)}") +_REPLY_LOCATION_PATTERN = re.compile( + r"\$message\.(header|payload)(#(\/(([^\/~\n])|(~[01]))*)*)?" +) P = ParamSpec("P") T = TypeVar("T") @@ -99,6 +102,20 @@ class Header(FieldInfo): # type: ignore[misc] pass +class ReplyTo(Header): # type: ignore[misc] + """ + Marks a header as carrying the address the reply should be sent to. + + It resolves like :py:class:`Header`, and additionally lets a channel declared with :py:class:`Reply` + derive the AsyncAPI reply address from the header, so the header name is only written once. It can be + used in a channel signature or in a dependency of the channel: + + .. code-block:: python + + reply_to: Annotated[str, ReplyTo(alias="replyTo")] + """ + + class Payload(FieldInfo): # type: ignore[misc] __schema_format__: ClassVar[str | None] = None @@ -117,6 +134,66 @@ class Parameter(FieldInfo): # type: ignore[misc] pass +@dataclass(frozen=True) +class ReplyAddress: + """ + A dynamically determined reply address, following the AsyncAPI Operation Reply Address object. + + :var location: + A runtime expression that specifies the location of the reply address, for example + "$message.header#/replyTo". It must follow the AsyncAPI runtime expression format, where either + "$message.header" or "$message.payload" is followed by a JSON Pointer. + :var description: An optional description of the reply address. + """ + + location: str + _: KW_ONLY + description: str | None = None + + def __post_init__(self) -> None: + if not _REPLY_LOCATION_PATTERN.fullmatch(self.location): + raise InvalidChannelDefinitionError( + "Reply address location must be a runtime expression of the form " + '"$message.header#/" or "$message.payload#/", ' + f"got: {self.location!r}" + ) + + +@dataclass(frozen=True) +class Reply: + """ + Marks a channel as implementing the request/reply pattern. + + The reply is the message sent by the channel handler, which must therefore send exactly one message + type. The AsyncAPI Operation Reply object is added to the receive operation of the channel, where the + reply channel is the channel of the sent message: + + .. code-block:: python + + @dataclass + class Pong(Message, address="pong"): + payload: str + + + @app.channel("ping", reply=Reply()) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") + + :var address: + An optional reply address, for when the address of the reply is only known at runtime, typically + supplied by the requester, for example in a replyTo header. A str location is converted to a + ReplyAddress. When omitted, the address is derived from a header marked with :py:class:`ReplyTo`, + if the channel has one. + """ + + _: KW_ONLY + address: ReplyAddress | str | None = None + + def __post_init__(self) -> None: + if isinstance(self.address, str): + object.__setattr__(self, "address", ReplyAddress(self.address)) + + @dataclass(frozen=True) class Depends: func: Callable[..., Any] @@ -189,6 +266,7 @@ class HeaderResolver(TypeResolve[T]): name: str default: T required: bool + reply_to: bool = False def resolve(self, message_receive: MessageReceive, send: AMGISendCallable) -> T: if not self.required: @@ -382,6 +460,7 @@ def resolve( class Channel(CallableResolver, ABC): address: str parameters: set[str] + reply: Reply | None = None async def __call__( self, @@ -509,6 +588,7 @@ def parameter_resolver( header_name, parameter.default, parameter.default is parameter.empty, + isinstance(annotation, ReplyTo), ) if isinstance(annotation, Binding): return BindingResolver( @@ -565,7 +645,9 @@ def resolvers_dependencies( return tuple(resolvers), tuple(dependencies) -def get_channel(func: Callable[..., Any], address: str) -> Channel: +def get_channel( + func: Callable[..., Any], address: str, reply: Reply | None = None +) -> Channel: address_parameters = get_address_parameters(address) resolvers, dependencies = resolvers_dependencies(func, address_parameters) @@ -574,16 +656,20 @@ def get_channel(func: Callable[..., Any], address: str) -> Channel: raise InvalidChannelDefinitionError("Channel must have no more than 1 payload") if inspect.iscoroutinefunction(func): - return AsyncChannel(func, resolvers, dependencies, address, address_parameters) + return AsyncChannel( + func, resolvers, dependencies, address, address_parameters, reply + ) if inspect.isasyncgenfunction(func): return AsyncGeneratorChannel( - func, resolvers, dependencies, address, address_parameters + func, resolvers, dependencies, address, address_parameters, reply ) if inspect.isgeneratorfunction(func): return SyncGeneratorChannel( - func, resolvers, dependencies, address, address_parameters + func, resolvers, dependencies, address, address_parameters, reply ) - return SyncChannel(func, resolvers, dependencies, address, address_parameters) + return SyncChannel( + func, resolvers, dependencies, address, address_parameters, reply + ) class ChannelRouter(Router[Channel]): @@ -591,8 +677,12 @@ class ChannelRouter(Router[Channel]): def channels(self) -> Sequence[Channel]: return [channel for _, channel in self.routes] - def add_channel(self, address: str, func: Callable[..., Any]) -> None: - self.add_route(address, get_channel(func, address)) + def add_channel( + self, address: str, func: Callable[..., Any], reply: Reply | None = None + ) -> Channel: + channel = get_channel(func, address, reply) + self.add_route(address, channel) + return channel async def __call__( self, scope: MessageScope, receive: AMGIReceiveCallable, send: AMGISendCallable diff --git a/packages/asyncfast/tests_asyncfast/test_asyncapi.py b/packages/asyncfast/tests_asyncfast/test_asyncapi.py index c3e6ee72..9a673156 100644 --- a/packages/asyncfast/tests_asyncfast/test_asyncapi.py +++ b/packages/asyncfast/tests_asyncfast/test_asyncapi.py @@ -5,12 +5,17 @@ from typing import Union from uuid import UUID +import pytest from asyncfast import AsyncFast from asyncfast import Depends from asyncfast import Header +from asyncfast import InvalidChannelDefinitionError from asyncfast import Message from asyncfast import MessageSender from asyncfast import Payload +from asyncfast import Reply +from asyncfast import ReplyAddress +from asyncfast import ReplyTo from asyncfast.bindings import KafkaKey from pydantic import BaseModel @@ -1219,3 +1224,368 @@ async def topic_handler(payload, name): # type: ignore[no-untyped-def] } }, } + + +def test_asyncapi_reply() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel("ping", reply=Reply()) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + assert app.asyncapi() == { + "asyncapi": "3.0.0", + "channels": { + "Ping": { + "address": "ping", + "messages": { + "PingMessage": {"$ref": "#/components/messages/PingMessage"} + }, + }, + "Pong": { + "address": "pong", + "messages": {"Pong": {"$ref": "#/components/messages/Pong"}}, + }, + }, + "components": { + "messages": { + "PingMessage": {"payload": {"type": "string"}}, + "Pong": {"payload": {"type": "string"}}, + } + }, + "info": {"title": "AsyncFast", "version": "0.1.0"}, + "operations": { + "receivePing": { + "action": "receive", + "channel": {"$ref": "#/channels/Ping"}, + "reply": { + "channel": {"$ref": "#/channels/Pong"}, + "messages": [{"$ref": "#/channels/Pong/messages/Pong"}], + }, + }, + "sendPong": {"action": "send", "channel": {"$ref": "#/channels/Pong"}}, + }, + } + + +def test_asyncapi_reply_address_string() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel("ping", reply=Reply(address="$message.header#/replyTo")) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + assert app.asyncapi() == { + "asyncapi": "3.0.0", + "channels": { + "Ping": { + "address": "ping", + "messages": { + "PingMessage": {"$ref": "#/components/messages/PingMessage"} + }, + }, + "Pong": { + "address": None, + "messages": {"Pong": {"$ref": "#/components/messages/Pong"}}, + }, + }, + "components": { + "messages": { + "PingMessage": {"payload": {"type": "string"}}, + "Pong": {"payload": {"type": "string"}}, + } + }, + "info": {"title": "AsyncFast", "version": "0.1.0"}, + "operations": { + "receivePing": { + "action": "receive", + "channel": {"$ref": "#/channels/Ping"}, + "reply": { + "address": {"location": "$message.header#/replyTo"}, + "channel": {"$ref": "#/channels/Pong"}, + "messages": [{"$ref": "#/channels/Pong/messages/Pong"}], + }, + }, + "sendPong": {"action": "send", "channel": {"$ref": "#/channels/Pong"}}, + }, + } + + +def test_asyncapi_reply_address_description() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel( + "ping", + reply=Reply( + address=ReplyAddress( + "$message.header#/replyTo", + description="Consumer inbox", + ) + ), + ) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + assert app.asyncapi() == { + "asyncapi": "3.0.0", + "channels": { + "Ping": { + "address": "ping", + "messages": { + "PingMessage": {"$ref": "#/components/messages/PingMessage"} + }, + }, + "Pong": { + "address": None, + "messages": {"Pong": {"$ref": "#/components/messages/Pong"}}, + }, + }, + "components": { + "messages": { + "PingMessage": {"payload": {"type": "string"}}, + "Pong": {"payload": {"type": "string"}}, + } + }, + "info": {"title": "AsyncFast", "version": "0.1.0"}, + "operations": { + "receivePing": { + "action": "receive", + "channel": {"$ref": "#/channels/Ping"}, + "reply": { + "address": { + "location": "$message.header#/replyTo", + "description": "Consumer inbox", + }, + "channel": {"$ref": "#/channels/Pong"}, + "messages": [{"$ref": "#/channels/Pong/messages/Pong"}], + }, + }, + "sendPong": {"action": "send", "channel": {"$ref": "#/channels/Pong"}}, + }, + } + + +def test_asyncapi_reply_same_channel() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="ping"): + payload: str + + @app.channel("ping", reply=Reply()) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + assert app.asyncapi() == { + "asyncapi": "3.0.0", + "channels": { + "Ping": { + "address": "ping", + "messages": { + "PingMessage": {"$ref": "#/components/messages/PingMessage"} + }, + }, + "Pong": { + "address": "ping", + "messages": {"Pong": {"$ref": "#/components/messages/Pong"}}, + }, + }, + "components": { + "messages": { + "PingMessage": {"payload": {"type": "string"}}, + "Pong": {"payload": {"type": "string"}}, + } + }, + "info": {"title": "AsyncFast", "version": "0.1.0"}, + "operations": { + "receivePing": { + "action": "receive", + "channel": {"$ref": "#/channels/Ping"}, + "reply": { + "channel": {"$ref": "#/channels/Pong"}, + "messages": [{"$ref": "#/channels/Pong/messages/Pong"}], + }, + }, + "sendPong": {"action": "send", "channel": {"$ref": "#/channels/Pong"}}, + }, + } + + +def test_asyncapi_reply_message_sender() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel("ping", reply=Reply()) + async def ping(message_sender: MessageSender[Pong]) -> None: + await message_sender.send(Pong(payload="pong")) # pragma: no cover + + assert app.asyncapi() == { + "asyncapi": "3.0.0", + "channels": { + "Ping": { + "address": "ping", + "messages": { + "PingMessage": {"$ref": "#/components/messages/PingMessage"} + }, + }, + "Pong": { + "address": "pong", + "messages": {"Pong": {"$ref": "#/components/messages/Pong"}}, + }, + }, + "components": { + "messages": { + "PingMessage": {}, + "Pong": {"payload": {"type": "string"}}, + } + }, + "info": {"title": "AsyncFast", "version": "0.1.0"}, + "operations": { + "receivePing": { + "action": "receive", + "channel": {"$ref": "#/channels/Ping"}, + "reply": { + "channel": {"$ref": "#/channels/Pong"}, + "messages": [{"$ref": "#/channels/Pong/messages/Pong"}], + }, + }, + "sendPong": {"action": "send", "channel": {"$ref": "#/channels/Pong"}}, + }, + } + + +def test_asyncapi_reply_dynamic_address_with_parameters() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="{reply_to}"): + reply_to: str + payload: str + + @app.channel("ping", reply=Reply(address=ReplyAddress("$message.header#/replyTo"))) + async def ping(payload: str) -> AsyncGenerator[Pong, None]: + yield Pong(reply_to="inbox", payload="pong") # pragma: no cover + + assert app.asyncapi()["channels"]["Pong"] == { + "address": None, + "messages": {"Pong": {"$ref": "#/components/messages/Pong"}}, + } + + +def test_asyncapi_reply_to_header_derives_address() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="{reply_to}"): + reply_to: str + payload: str + + @app.channel("ping", reply=Reply()) + async def ping( + reply_to: Annotated[str, ReplyTo(alias="replyTo")], + ) -> AsyncGenerator[Pong, None]: + yield Pong(reply_to=reply_to, payload="pong") # pragma: no cover + + schema = app.asyncapi() + assert schema["operations"]["receivePing"]["reply"]["address"] == { + "location": "$message.header#/replyTo" + } + assert schema["channels"]["Pong"]["address"] is None + assert "headers" in schema["components"]["messages"]["PingMessage"] + + +def test_asyncapi_reply_to_header_via_dependency() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="{reply_to}"): + reply_to: str + payload: str + + def get_reply_to(reply_to: Annotated[str, ReplyTo()]) -> str: + return reply_to # pragma: no cover + + @app.channel("ping", reply=Reply()) + async def ping( + reply_to: Annotated[str, Depends(get_reply_to)], + ) -> AsyncGenerator[Pong, None]: + yield Pong(reply_to=reply_to, payload="pong") # pragma: no cover + + assert app.asyncapi()["operations"]["receivePing"]["reply"]["address"] == { + "location": "$message.header#/reply-to" + } + + +def test_asyncapi_explicit_reply_address_wins_over_reply_to() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel("ping", reply=Reply(address="$message.header#/other")) + async def ping( + reply_to: Annotated[str, ReplyTo()], + ) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + assert app.asyncapi()["operations"]["receivePing"]["reply"]["address"] == { + "location": "$message.header#/other" + } + + +def test_asyncapi_reply_to_without_reply_is_plain_header() -> None: + app = AsyncFast() + + @app.channel("ping") + async def ping(reply_to: Annotated[str, ReplyTo()]) -> None: + pass # pragma: no cover + + assert "reply" not in app.asyncapi()["operations"]["receivePing"] + + +def test_asyncapi_reply_without_reply_to_has_no_address() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel("ping", reply=Reply()) + async def ping() -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + assert "address" not in app.asyncapi()["operations"]["receivePing"]["reply"] + + +def test_asyncapi_multiple_reply_to_headers_raises() -> None: + app = AsyncFast() + + @dataclass + class Pong(Message, address="pong"): + payload: str + + @app.channel("ping", reply=Reply()) + async def ping( + first: Annotated[str, ReplyTo()], + second: Annotated[str, ReplyTo()], + ) -> AsyncGenerator[Pong, None]: + yield Pong(payload="pong") # pragma: no cover + + with pytest.raises(InvalidChannelDefinitionError): + app.asyncapi() diff --git a/packages/asyncfast/tests_asyncfast/test_errors.py b/packages/asyncfast/tests_asyncfast/test_errors.py index d10d5b28..cf0d043e 100644 --- a/packages/asyncfast/tests_asyncfast/test_errors.py +++ b/packages/asyncfast/tests_asyncfast/test_errors.py @@ -1,6 +1,15 @@ +from collections.abc import AsyncGenerator +from dataclasses import dataclass +from typing import Union + import pytest from asyncfast import AsyncFast from asyncfast import InvalidChannelDefinitionError +from asyncfast import Message +from asyncfast import Reply +from asyncfast import ReplyAddress +from asyncfast._asyncapi import get_asyncapi +from asyncfast._channel import ChannelRouter async def test_only_one_payload() -> None: @@ -11,3 +20,102 @@ async def test_only_one_payload() -> None: @app.channel("topic") async def topic_handler(payload1: int, payload2: int) -> None: pass # pragma: no cover + + +def test_reply_address_invalid_location() -> None: + with pytest.raises(InvalidChannelDefinitionError): + ReplyAddress("replyTo") + + +def test_reply_address_location_without_fragment() -> None: + assert ReplyAddress("$message.header").location == "$message.header" + assert ReplyAddress("$message.payload").location == "$message.payload" + + +def test_reply_address_empty_fragment_location() -> None: + assert ReplyAddress("$message.header#").location == "$message.header#" + + +def test_reply_address_location_with_escapes() -> None: + location = "$message.payload#/a~1b~0c" + assert ReplyAddress(location).location == location + + +def test_reply_address_trailing_newline_location() -> None: + with pytest.raises(InvalidChannelDefinitionError): + ReplyAddress("$message.header#/replyTo\n") + + +def test_reply_address_invalid_escape_location() -> None: + with pytest.raises(InvalidChannelDefinitionError): + ReplyAddress("$message.header#/replyTo~2") + + +def test_reply_address_invalid_reference_location() -> None: + with pytest.raises(InvalidChannelDefinitionError): + ReplyAddress("$message.headers#/replyTo") + + +def test_reply_invalid_str_address() -> None: + with pytest.raises(InvalidChannelDefinitionError): + Reply(address="not-a-runtime-expression") + + +def test_reply_no_send_message() -> None: + app = AsyncFast() + + with pytest.raises(InvalidChannelDefinitionError): + + @app.channel("ping", reply=Reply()) + async def ping(payload: int) -> None: + pass # pragma: no cover + + +def test_reply_multiple_send_messages() -> None: + @dataclass + class Pong(Message, address="pong"): + payload: str + + @dataclass + class PongError(Message, address="pong_error"): + payload: str + + app = AsyncFast() + + with pytest.raises(InvalidChannelDefinitionError): + + @app.channel("ping", reply=Reply()) + async def ping(payload: int) -> AsyncGenerator[Union[Pong, PongError], None]: + yield Pong(payload="pong") # pragma: no cover + + +def test_reply_invalid_channel_not_registered() -> None: + app = AsyncFast() + + with pytest.raises(InvalidChannelDefinitionError): + + @app.channel("ping", reply=Reply()) + async def ping(payload: int) -> None: + pass # pragma: no cover + + assert app._router.channels == [] + assert app.asyncapi()["channels"] == {} + + +def test_reply_multiple_send_messages_asyncapi() -> None: + @dataclass + class Pong(Message, address="pong"): + payload: str + + @dataclass + class PongError(Message, address="pong_error"): + payload: str + + async def ping(payload: int) -> AsyncGenerator[Union[Pong, PongError], None]: + yield Pong(payload="pong") # pragma: no cover + + router = ChannelRouter() + router.add_channel("ping", ping, Reply()) + + with pytest.raises(InvalidChannelDefinitionError): + get_asyncapi(title="AsyncFast", version="0.1.0", router=router)