Merge pull request #433 from Clariteia/0.1.1

0.1.1

Author: Sergio García Prado

HISTORY.md

@@ -120,3 +120,9 @@ History
 ------------------
 
 * Add `minos-microservice-common>=0.2.0` compatibility.
+
+0.1.1 (2021-11-09)
+------------------
+
+* Add `REPLY_TOPIC_CONTEXT_VAR` and integrate with `DynamicHandlerPool`.
+* Add support for `post_fn` callbacks following the same strategy as in `pre_fn` callbacks.

minos/networks/__init__.py

@@ -1,8 +1,9 @@
 __author__ = """Clariteia Devs"""
 __email__ = "devs@clariteia.com"
-__version__ = "0.1.0"
+__version__ = "0.1.1"
 
 from .brokers import (
+    REPLY_TOPIC_CONTEXT_VAR,
     Broker,
     BrokerSetup,
     CommandBroker,

minos/networks/brokers/__init__.py

@@ -6,6 +6,7 @@
     CommandReplyBroker,
 )
 from .commands import (
+    REPLY_TOPIC_CONTEXT_VAR,
     CommandBroker,
 )
 from .events import (

minos/networks/brokers/commands.py

@@ -3,8 +3,12 @@
 )
 
 import logging
+from contextvars import (
+    ContextVar,
+)
 from typing import (
     Any,
+    Final,
     Optional,
 )
 from uuid import (
@@ -22,6 +26,8 @@
 
 logger = logging.getLogger(__name__)
 
+REPLY_TOPIC_CONTEXT_VAR: Final[ContextVar[Optional[str]]] = ContextVar("reply_topic", default=None)
+
 
 class CommandBroker(Broker):
     """Minos Command Broker Class."""
@@ -57,8 +63,11 @@ async def send(
         the command is not authenticated.
         :return: This method does not return anything.
         """
+        if reply_topic is None:
+            reply_topic = REPLY_TOPIC_CONTEXT_VAR.get()
         if reply_topic is None:
             reply_topic = self.default_reply_topic
+
         command = Command(topic, data, saga, reply_topic, user)
         logger.info(f"Sending '{command!s}'...")
         return await self.enqueue(command.topic, command.avro_bytes)

minos/networks/decorators/builders.py

@@ -5,7 +5,7 @@
     defaultdict,
 )
 from inspect import (
-    iscoroutinefunction,
+    isawaitable,
 )
 from typing import (
     Awaitable,
@@ -112,34 +112,30 @@ def _build_one_class(
 
         for name, decorators in mapping.items():
             for decorator in decorators:
-                ans[decorator].add(self._build_one_method(class_, name, decorator.pre_fn_name))
+                ans[decorator].add(self._build_one_method(class_, name, decorator.pre_fn_name, decorator.post_fn_name))
 
     @staticmethod
-    def _build_one_method(class_: type, name: str, pref_fn_name: str, **kwargs) -> Handler:
+    def _build_one_method(class_: type, name: str, pref_fn_name: str, post_fn_name: str, **kwargs) -> Handler:
         instance = class_(**kwargs)
         fn = getattr(instance, name)
         pre_fn = getattr(instance, pref_fn_name, None)
+        post_fn = getattr(instance, post_fn_name, None)
 
-        if iscoroutinefunction(fn):
-            _awaitable_fn = fn
-        else:
+        async def _wrapped_fn(request: Request) -> Optional[Response]:
+            if pre_fn is not None:
+                request = pre_fn(request)
+                if isawaitable(request):
+                    request = await request
 
-            async def _awaitable_fn(request: Request) -> Optional[Response]:
-                return fn(request)
+            response = fn(request)
+            if isawaitable(response):
+                response = await response
 
-        if pre_fn is None:
-            _wrapped_fn = _awaitable_fn
-        else:
-            if iscoroutinefunction(pre_fn):
+            if post_fn is not None:
+                response = post_fn(response)
+                if isawaitable(response):
+                    response = await response
 
-                async def _wrapped_fn(request: Request) -> Optional[Response]:
-                    request = await pre_fn(request)
-                    return await _awaitable_fn(request)
-
-            else:
-
-                async def _wrapped_fn(request: Request) -> Optional[Response]:
-                    request = pre_fn(request)
-                    return await _awaitable_fn(request)
+            return response
 
         return _wrapped_fn

minos/networks/decorators/definitions/abc.py

@@ -81,3 +81,11 @@ def pre_fn_name(self) -> str:
         :return: A string value containing the function name.
         """
         return self.KIND.pre_fn_name
+
+    @property
+    def post_fn_name(self) -> str:
+        """Get the post execution function name.
+
+        :return: A string value containing the function name.
+        """
+        return self.KIND.post_fn_name

minos/networks/decorators/definitions/kinds.py

@@ -23,3 +23,16 @@ def pre_fn_name(self) -> str:
             self.Event: "_pre_event_handle",
         }
         return mapping[self]
+
+    @property
+    def post_fn_name(self) -> str:
+        """Get the post execution function name.
+
+        :return: A string value containing the function name.
+        """
+        mapping = {
+            self.Command: "_post_command_handle",
+            self.Query: "_post_query_handle",
+            self.Event: "_post_event_handle",
+        }
+        return mapping[self]

minos/networks/handlers/dynamic/pools.py

@@ -3,6 +3,13 @@
 )
 
 import logging
+from contextvars import (
+    Token,
+)
+from typing import (
+    AsyncContextManager,
+    Optional,
+)
 from uuid import (
     uuid4,
 )
@@ -23,6 +30,9 @@
     MinosPool,
 )
 
+from ...brokers import (
+    REPLY_TOPIC_CONTEXT_VAR,
+)
 from ..consumers import (
     Consumer,
 )
@@ -82,3 +92,27 @@ async def _subscribe_reply_topic(self, topic: str) -> None:
 
     async def _unsubscribe_reply_topic(self, topic: str) -> None:
         await self.consumer.remove_topic(topic)
+
+    def acquire(self, *args, **kwargs) -> AsyncContextManager:
+        """Acquire a new instance wrapped on an asynchronous context manager.
+
+        :return: An asynchronous context manager.
+        """
+        return _ReplyTopicContextManager(super().acquire())
+
+
+class _ReplyTopicContextManager:
+    _token: Optional[Token]
+
+    def __init__(self, wrapper: AsyncContextManager[DynamicHandler]):
+        self.wrapper = wrapper
+        self._token = None
+
+    async def __aenter__(self) -> DynamicHandler:
+        handler = await self.wrapper.__aenter__()
+        self._token = REPLY_TOPIC_CONTEXT_VAR.set(handler.topic)
+        return handler
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        REPLY_TOPIC_CONTEXT_VAR.reset(self._token)
+        await self.wrapper.__aexit__(exc_type, exc_val, exc_tb)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "minos_microservice_networks"
-version = "0.1.0"
+version = "0.1.1"
 description = "Python Package with the common network classes and utilities used in Minos Microservice."
 readme = "README.md"
 repository = "https://github.com/clariteia/minos_microservice_network"

tests/test_networks/test_brokers/test_commands.py

@@ -13,6 +13,7 @@
     PostgresAsyncTestCase,
 )
 from minos.networks import (
+    REPLY_TOPIC_CONTEXT_VAR,
     CommandBroker,
 )
 from tests.utils import (
@@ -65,6 +66,23 @@ async def test_send_with_default_reply_topic(self):
         self.assertEqual("fake", args[0])
         self.assertEqual(Command("fake", FakeModel("foo"), saga, "OrderReply"), Command.from_avro_bytes(args[1]))
 
+    async def test_send_with_reply_topic_context_var(self):
+        mock = AsyncMock(return_value=56)
+        saga = uuid4()
+
+        REPLY_TOPIC_CONTEXT_VAR.set("onetwothree")
+
+        async with CommandBroker.from_config(config=self.config) as broker:
+            broker.enqueue = mock
+            identifier = await broker.send(FakeModel("foo"), "fake", saga)
+
+        self.assertEqual(56, identifier)
+        self.assertEqual(1, mock.call_count)
+
+        args = mock.call_args.args
+        self.assertEqual("fake", args[0])
+        self.assertEqual(Command("fake", FakeModel("foo"), saga, "onetwothree"), Command.from_avro_bytes(args[1]))
+
     async def test_send_with_user(self):
         mock = AsyncMock(return_value=56)
         saga = uuid4()