Fix middlewares order (#1935)

* Fix middlewares order for publishing scope * Missing middlewares * ruff * add tests * ruff tests * pre-commit * Add test for `aenter`, `aexit` * pre-commit * Update tests * Fix tests * Fix tests again * lint: fix mypy * Update tests * pre-commit * Fix tests * Update tests, again :) * format tests * Remove logger from tests * format * fix: correct NATS Publisher middlewares order * fix: correct All Publisher middlewares order * Try to fix types * format * chore: fix CI * fix: correct confluent middlewares order * tests: make order middlewares tests in-memory --------- Co-authored-by: Nikita Pastukhov <[email protected]> Co-authored-by: Pastukhov Nikita <[email protected]>
airtai · Nov 29, 2024 · a173bb5 · a173bb5
1 parent 90aaf57
commit a173bb5
Show file tree

Hide file tree

Showing 58 changed files with 685 additions and 266 deletions.
diff --git a/faststream/broker/core/abc.py b/faststream/broker/core/abc.py
@@ -6,6 +6,7 @@
     Iterable,
     Mapping,
     Optional,
+    Sequence,
 )
 
 from faststream.broker.types import MsgType
@@ -30,7 +31,7 @@ def __init__(
         *,
         prefix: str,
         dependencies: Iterable["Depends"],
-        middlewares: Iterable["BrokerMiddleware[MsgType]"],
+        middlewares: Sequence["BrokerMiddleware[MsgType]"],
         parser: Optional["CustomCallable"],
         decoder: Optional["CustomCallable"],
         include_in_schema: Optional[bool],

diff --git a/faststream/broker/core/usecase.py b/faststream/broker/core/usecase.py
@@ -75,7 +75,7 @@ def __init__(
             Doc("Dependencies to apply to all broker subscribers."),
         ],
         middlewares: Annotated[
-            Iterable["BrokerMiddleware[MsgType]"],
+            Sequence["BrokerMiddleware[MsgType]"],
             Doc("Middlewares to apply to all broker publishers/subscribers."),
         ],
         graceful_timeout: Annotated[
@@ -342,7 +342,7 @@ async def publish(
 
         publish = producer.publish
 
-        for m in self._middlewares:
+        for m in self._middlewares[::-1]:
             publish = partial(m(None).publish_scope, publish)
 
         return await publish(msg, correlation_id=correlation_id, **kwargs)
@@ -359,7 +359,7 @@ async def request(
         assert producer, NOT_CONNECTED_YET  # nosec B101
 
         request = producer.request
-        for m in self._middlewares:
+        for m in self._middlewares[::-1]:
             request = partial(m(None).publish_scope, request)
 
         published_msg = await request(
@@ -370,7 +370,7 @@ async def request(
 
         async with AsyncExitStack() as stack:
             return_msg = return_input
-            for m in self._middlewares:
+            for m in self._middlewares[::-1]:
                 mid = m(published_msg)
                 await stack.enter_async_context(mid)
                 return_msg = partial(mid.consume_scope, return_msg)

diff --git a/faststream/broker/fastapi/router.py b/faststream/broker/fastapi/router.py
@@ -105,7 +105,7 @@ class StreamRouter(
     def __init__(
         self,
         *connection_args: Any,
-        middlewares: Iterable["BrokerMiddleware[MsgType]"] = (),
+        middlewares: Sequence["BrokerMiddleware[MsgType]"] = (),
         prefix: str = "",
         tags: Optional[List[Union[str, Enum]]] = None,
         dependencies: Optional[Sequence["params.Depends"]] = None,

diff --git a/faststream/broker/publisher/fake.py b/faststream/broker/publisher/fake.py
@@ -1,6 +1,6 @@
 from functools import partial
 from itertools import chain
-from typing import TYPE_CHECKING, Any, Iterable, Optional
+from typing import TYPE_CHECKING, Any, Optional, Sequence
 
 from faststream.broker.publisher.proto import BasePublisherProto
 
@@ -17,7 +17,7 @@ def __init__(
         method: "AsyncFunc",
         *,
         publish_kwargs: "AnyDict",
-        middlewares: Iterable["PublisherMiddleware"] = (),
+        middlewares: Sequence["PublisherMiddleware"] = (),
     ) -> None:
         """Initialize an object."""
         self.method = method
@@ -29,7 +29,7 @@ async def publish(
         message: "SendableMessage",
         *,
         correlation_id: Optional[str] = None,
-        _extra_middlewares: Iterable["PublisherMiddleware"] = (),
+        _extra_middlewares: Sequence["PublisherMiddleware"] = (),
         **kwargs: Any,
     ) -> Any:
         """Publish a message."""
@@ -51,7 +51,7 @@ async def request(
         /,
         *,
         correlation_id: Optional[str] = None,
-        _extra_middlewares: Iterable["PublisherMiddleware"] = (),
+        _extra_middlewares: Sequence["PublisherMiddleware"] = (),
     ) -> Any:
         raise NotImplementedError(
             "`FakePublisher` can be used only to publish "

diff --git a/faststream/broker/publisher/proto.py b/faststream/broker/publisher/proto.py
@@ -1,5 +1,13 @@
 from abc import abstractmethod
-from typing import TYPE_CHECKING, Any, Callable, Generic, Iterable, Optional, Protocol
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Generic,
+    Optional,
+    Protocol,
+    Sequence,
+)
 
 from typing_extensions import override
 
@@ -53,7 +61,7 @@ async def publish(
         /,
         *,
         correlation_id: Optional[str] = None,
-        _extra_middlewares: Iterable["PublisherMiddleware"] = (),
+        _extra_middlewares: Sequence["PublisherMiddleware"] = (),
     ) -> Optional[Any]:
         """Publishes a message asynchronously."""
         ...
@@ -65,7 +73,7 @@ async def request(
         /,
         *,
         correlation_id: Optional[str] = None,
-        _extra_middlewares: Iterable["PublisherMiddleware"] = (),
+        _extra_middlewares: Sequence["PublisherMiddleware"] = (),
     ) -> Optional[Any]:
         """Publishes a message synchronously."""
         ...
@@ -79,8 +87,8 @@ class PublisherProto(
 ):
     schema_: Any
 
-    _broker_middlewares: Iterable["BrokerMiddleware[MsgType]"]
-    _middlewares: Iterable["PublisherMiddleware"]
+    _broker_middlewares: Sequence["BrokerMiddleware[MsgType]"]
+    _middlewares: Sequence["PublisherMiddleware"]
     _producer: Optional["ProducerProto"]
 
     @abstractmethod

diff --git a/faststream/broker/publisher/usecase.py b/faststream/broker/publisher/usecase.py
@@ -4,9 +4,9 @@
     TYPE_CHECKING,
     Any,
     Callable,
-    Iterable,
     List,
     Optional,
+    Sequence,
     Tuple,
 )
 from unittest.mock import MagicMock
@@ -49,11 +49,11 @@ def __init__(
         self,
         *,
         broker_middlewares: Annotated[
-            Iterable["BrokerMiddleware[MsgType]"],
+            Sequence["BrokerMiddleware[MsgType]"],
             Doc("Top-level middlewares to use in direct `.publish` call."),
         ],
         middlewares: Annotated[
-            Iterable["PublisherMiddleware"],
+            Sequence["PublisherMiddleware"],
             Doc("Publisher middlewares."),
         ],
         # AsyncAPI args

diff --git a/faststream/broker/router.py b/faststream/broker/router.py
@@ -4,6 +4,7 @@
     Callable,
     Iterable,
     Optional,
+    Sequence,
 )
 
 from faststream.broker.core.abc import ABCBroker
@@ -64,7 +65,7 @@ def __init__(
         # base options
         prefix: str,
         dependencies: Iterable["Depends"],
-        middlewares: Iterable["BrokerMiddleware[MsgType]"],
+        middlewares: Sequence["BrokerMiddleware[MsgType]"],
         parser: Optional["CustomCallable"],
         decoder: Optional["CustomCallable"],
         include_in_schema: Optional[bool],

diff --git a/faststream/broker/subscriber/call_item.py b/faststream/broker/subscriber/call_item.py
@@ -9,6 +9,7 @@
     Generic,
     Iterable,
     Optional,
+    Sequence,
     cast,
 )
 
@@ -54,7 +55,7 @@ def __init__(
         filter: "AsyncFilter[StreamMessage[MsgType]]",
         item_parser: Optional["CustomCallable"],
         item_decoder: Optional["CustomCallable"],
-        item_middlewares: Iterable["SubscriberMiddleware[StreamMessage[MsgType]]"],
+        item_middlewares: Sequence["SubscriberMiddleware[StreamMessage[MsgType]]"],
         dependencies: Iterable["Depends"],
     ) -> None:
         self.handler = handler
@@ -157,7 +158,7 @@ async def call(
         """Execute wrapped handler with consume middlewares."""
         call: AsyncFuncAny = self.handler.call_wrapped
 
-        for middleware in chain(self.item_middlewares, _extra_middlewares):
+        for middleware in chain(self.item_middlewares[::-1], _extra_middlewares):
             call = partial(middleware, call)
 
         try:

diff --git a/faststream/broker/subscriber/proto.py b/faststream/broker/subscriber/proto.py
@@ -1,5 +1,14 @@
 from abc import abstractmethod
-from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Sequence,
+)
 
 from typing_extensions import Self, override
 
@@ -33,7 +42,7 @@ class SubscriberProto(
     running: bool
 
     _broker_dependencies: Iterable["Depends"]
-    _broker_middlewares: Iterable["BrokerMiddleware[MsgType]"]
+    _broker_middlewares: Sequence["BrokerMiddleware[MsgType]"]
     _producer: Optional["ProducerProto"]
 
     @abstractmethod
@@ -98,6 +107,6 @@ def add_call(
         filter_: "Filter[Any]",
         parser_: "CustomCallable",
         decoder_: "CustomCallable",
-        middlewares_: Iterable["SubscriberMiddleware[Any]"],
+        middlewares_: Sequence["SubscriberMiddleware[Any]"],
         dependencies_: Iterable["Depends"],
     ) -> Self: ...
diff --git a/faststream/broker/subscriber/usecase.py b/faststream/broker/subscriber/usecase.py
@@ -10,6 +10,7 @@
     Iterable,
     List,
     Optional,
+    Sequence,
     Tuple,
     Union,
     overload,
@@ -53,11 +54,11 @@
 
 class _CallOptions:
     __slots__ = (
-        "filter",
-        "parser",
         "decoder",
-        "middlewares",
         "dependencies",
+        "filter",
+        "middlewares",
+        "parser",
     )
 
     def __init__(
@@ -66,7 +67,7 @@ def __init__(
         filter: "Filter[Any]",
         parser: Optional["CustomCallable"],
         decoder: Optional["CustomCallable"],
-        middlewares: Iterable["SubscriberMiddleware[Any]"],
+        middlewares: Sequence["SubscriberMiddleware[Any]"],
         dependencies: Iterable["Depends"],
     ) -> None:
         self.filter = filter
@@ -98,7 +99,7 @@ def __init__(
         no_reply: bool,
         retry: Union[bool, int],
         broker_dependencies: Iterable["Depends"],
-        broker_middlewares: Iterable["BrokerMiddleware[MsgType]"],
+        broker_middlewares: Sequence["BrokerMiddleware[MsgType]"],
         default_parser: "AsyncCallable",
         default_decoder: "AsyncCallable",
         # AsyncAPI information
@@ -211,7 +212,7 @@ def add_call(
         filter_: "Filter[Any]",
         parser_: Optional["CustomCallable"],
         decoder_: Optional["CustomCallable"],
-        middlewares_: Iterable["SubscriberMiddleware[Any]"],
+        middlewares_: Sequence["SubscriberMiddleware[Any]"],
         dependencies_: Iterable["Depends"],
     ) -> Self:
         self._call_options = _CallOptions(
@@ -231,7 +232,7 @@ def __call__(
         filter: Optional["Filter[Any]"] = None,
         parser: Optional["CustomCallable"] = None,
         decoder: Optional["CustomCallable"] = None,
-        middlewares: Iterable["SubscriberMiddleware[Any]"] = (),
+        middlewares: Sequence["SubscriberMiddleware[Any]"] = (),
         dependencies: Iterable["Depends"] = (),
     ) -> Callable[
         [Callable[P_HandlerParams, T_HandlerReturn]],
@@ -246,7 +247,7 @@ def __call__(
         filter: Optional["Filter[Any]"] = None,
         parser: Optional["CustomCallable"] = None,
         decoder: Optional["CustomCallable"] = None,
-        middlewares: Iterable["SubscriberMiddleware[Any]"] = (),
+        middlewares: Sequence["SubscriberMiddleware[Any]"] = (),
         dependencies: Iterable["Depends"] = (),
     ) -> "HandlerCallWrapper[MsgType, P_HandlerParams, T_HandlerReturn]": ...
 
@@ -257,7 +258,7 @@ def __call__(
         filter: Optional["Filter[Any]"] = None,
         parser: Optional["CustomCallable"] = None,
         decoder: Optional["CustomCallable"] = None,
-        middlewares: Iterable["SubscriberMiddleware[Any]"] = (),
+        middlewares: Sequence["SubscriberMiddleware[Any]"] = (),
         dependencies: Iterable["Depends"] = (),
     ) -> Any:
         if (options := self._call_options) is None:
@@ -367,7 +368,9 @@ async def process_message(self, msg: MsgType) -> "Response":
                         await h.call(
                             message=message,
                             # consumer middlewares
-                            _extra_middlewares=(m.consume_scope for m in middlewares),
+                            _extra_middlewares=(
+                                m.consume_scope for m in middlewares[::-1]
+                            ),
                         )
                     )
 
@@ -382,7 +385,9 @@ async def process_message(self, msg: MsgType) -> "Response":
                             result_msg.body,
                             **result_msg.as_publish_kwargs(),
                             # publisher middlewares
-                            _extra_middlewares=(m.publish_scope for m in middlewares),
+                            _extra_middlewares=[
+                                m.publish_scope for m in middlewares[::-1]
+                            ],
                         )
 
                     # Return data for tests

diff --git a/faststream/broker/utils.py b/faststream/broker/utils.py
@@ -8,8 +8,8 @@
     AsyncContextManager,
     Awaitable,
     Callable,
-    Iterable,
     Optional,
+    Sequence,
     Type,
     Union,
     cast,
@@ -37,7 +37,7 @@
 
 async def process_msg(
     msg: Optional[MsgType],
-    middlewares: Iterable["BrokerMiddleware[MsgType]"],
+    middlewares: Sequence["BrokerMiddleware[MsgType]"],
     parser: Callable[[MsgType], Awaitable["StreamMessage[MsgType]"]],
     decoder: Callable[["StreamMessage[MsgType]"], "Any"],
 ) -> Optional["StreamMessage[MsgType]"]:
@@ -50,7 +50,7 @@ async def process_msg(
             Awaitable[StreamMessage[MsgType]],
         ] = return_input
 
-        for m in middlewares:
+        for m in middlewares[::-1]:
             mid = m(msg)
             await stack.enter_async_context(mid)
             return_msg = partial(mid.consume_scope, return_msg)