Merge pull request #420 from Clariteia/0.0.19

0.0.19

Author: Sergio García Prado

HISTORY.md

@@ -109,3 +109,9 @@ History
 
 * Add `PeriodicTask`, `PeriodicTaskScheduler` and `PeriodicTaskSchedulerService`.
 * Add `@enroute.periodic.event` decorator
+
+0.0.19 (2021-11-03)
+------------------
+
+* Add `"user"` context variable to be accessible during `Request` handling (same as `Request.user`).
+* Add support for `Request.user` propagation over `CommandBroker`.

minos/networks/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.0.18"
+__version__ = "0.0.19"
 
 from .brokers import (
     Broker,
@@ -60,6 +60,7 @@
     HandlerSetup,
 )
 from .messages import (
+    USER_CONTEXT_VAR,
     Request,
     Response,
     ResponseException,

minos/networks/brokers/commands.py

@@ -38,17 +38,27 @@ def _from_config(cls, *args, config: MinosConfig, **kwargs) -> CommandBroker:
         return cls(*args, **config.broker.queue._asdict(), default_reply_topic=default_reply_topic, **kwargs)
 
     # noinspection PyMethodOverriding
-    async def send(self, data: Any, topic: str, saga: UUID, reply_topic: Optional[str] = None, **kwargs) -> int:
+    async def send(
+        self,
+        data: Any,
+        topic: str,
+        saga: UUID,
+        reply_topic: Optional[str] = None,
+        user: Optional[UUID] = None,
+        **kwargs,
+    ) -> int:
         """Send a ``Command``.
 
         :param data: The data to be send.
         :param topic: Topic in which the message will be published.
         :param saga: Saga identifier.
         :param reply_topic: Topic name in which the reply will be published.
+        :param user: Optional user identifier. If the value is not `None` then the command is authenticated, otherwise
+        the command is not authenticated.
         :return: This method does not return anything.
         """
         if reply_topic is None:
             reply_topic = self.default_reply_topic
-        command = Command(topic, data, saga, 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/handlers/commands/handlers.py

@@ -31,6 +31,7 @@
     EnrouteBuilder,
 )
 from ...messages import (
+    USER_CONTEXT_VAR,
     Response,
     ResponseException,
 )
@@ -95,8 +96,10 @@ def get_callback(
         """
 
         async def _fn(command: Command) -> Tuple[Any, CommandStatus]:
+            request = HandlerRequest(command)
+            token = USER_CONTEXT_VAR.set(request.user)
+
             try:
-                request = HandlerRequest(command)
                 response = fn(request)
                 if isawaitable(response):
                     response = await response
@@ -109,5 +112,7 @@ async def _fn(command: Command) -> Tuple[Any, CommandStatus]:
             except Exception as exc:
                 logger.exception(f"Raised a system exception: {exc!r}")
                 return repr(exc), CommandStatus.SYSTEM_ERROR
+            finally:
+                USER_CONTEXT_VAR.reset(token)
 
         return _fn

minos/networks/messages.py

@@ -6,12 +6,16 @@
     ABC,
     abstractmethod,
 )
+from contextvars import (
+    ContextVar,
+)
 from inspect import (
     isawaitable,
 )
 from typing import (
     Any,
     Callable,
+    Final,
     Optional,
 )
 from uuid import (
@@ -26,6 +30,9 @@
     MinosException,
 )
 
+USER_CONTEXT_VAR: Final[ContextVar[Optional[UUID]]] = ContextVar("user", default=None)
+USER_CONTEXT_VAR.set(None)  # needed to "register" the context variable.
+
 
 class Request(ABC):
     """Request interface."""

minos/networks/rest/handlers.py

@@ -29,6 +29,7 @@
     EnrouteBuilder,
 )
 from ..messages import (
+    USER_CONTEXT_VAR,
     Response,
     ResponseException,
 )
@@ -127,6 +128,7 @@ async def _fn(request: web.Request) -> web.Response:
             logger.info(f"Dispatching '{request!s}' from '{request.remote!s}'...")
 
             request = RestRequest(request)
+            token = USER_CONTEXT_VAR.set(request.user)
 
             try:
                 response = fn(request)
@@ -143,6 +145,8 @@ async def _fn(request: web.Request) -> web.Response:
             except Exception as exc:
                 logger.exception(f"Raised a system exception: {exc!r}")
                 raise web.HTTPInternalServerError()
+            finally:
+                USER_CONTEXT_VAR.reset(token)
 
         return _fn
 

minos/networks/rest/messages.py

@@ -57,7 +57,18 @@ def user(self) -> Optional[UUID]:
         """
         Returns the UUID of the user making the Request.
         """
-        return UUID(self.raw_request.headers["User"])
+        if "User" not in self.headers:
+            return None
+        return UUID(self.headers["User"])
+
+    @property
+    def headers(self) -> dict[str, str]:
+        """Get the headers of the request.
+
+        :return: A dictionary in which keys are ``str`` instances and values are ``str`` instances.
+        """
+        # noinspection PyTypeChecker
+        return self.raw_request.headers
 
     async def content(self, model_type: Union[ModelType, Type[Model], str] = "Content", **kwargs) -> Any:
         """Get the request content.

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "minos_microservice_networks"
-version = "0.0.18"
+version = "0.0.19"
 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

@@ -65,6 +65,22 @@ 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_user(self):
+        mock = AsyncMock(return_value=56)
+        saga = uuid4()
+        user = uuid4()
+
+        async with CommandBroker.from_config(config=self.config) as broker:
+            broker.enqueue = mock
+            identifier = await broker.send(FakeModel("foo"), "fake", saga, "ekaf", user)
+
+        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, "ekaf", user), Command.from_avro_bytes(args[1]))
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_networks/test_handlers/test_commands/test_handlers.py

@@ -16,6 +16,7 @@
     PostgresAsyncTestCase,
 )
 from minos.networks import (
+    USER_CONTEXT_VAR,
     CommandHandler,
     HandlerEntry,
     HandlerRequest,
@@ -56,7 +57,8 @@ def setUp(self) -> None:
         super().setUp()
         self.broker = FakeBroker()
         self.handler = CommandHandler.from_config(config=self.config, broker=self.broker)
-        self.command = Command("AddOrder", FakeModel("foo"), uuid4(), "UpdateTicket")
+        self.user = uuid4()
+        self.command = Command("AddOrder", FakeModel("foo"), self.user, "UpdateTicket")
 
     def test_from_config(self):
         broker = FakeBroker()
@@ -117,6 +119,18 @@ async def test_get_callback_raises_exception(self):
         expected = (repr(ValueError()), CommandStatus.SYSTEM_ERROR)
         self.assertEqual(expected, await fn(self.command))
 
+    async def test_get_callback_with_user(self):
+        async def _fn(request) -> None:
+            self.assertEqual(self.user, request.user)
+            self.assertEqual(self.user, USER_CONTEXT_VAR.get())
+
+        mock = AsyncMock(side_effect=_fn)
+
+        handler = self.handler.get_callback(mock)
+        await handler(self.command)
+
+        self.assertEqual(1, mock.call_count)
+
 
 if __name__ == "__main__":
     unittest.main()