#23 refactor for cleverBRAG, fix failing unit tests

Reviewed-on: #44

.flake8

@@ -0,0 +1,4 @@
+[flake8]
+max-line-length = 88
+ignore = E501, W503, E203
+exclude = .git,__pycache__,docs/source/conf.py,old,build,dist,.venv

.pre-commit-config.yaml

@@ -0,0 +1,21 @@
+repos:
+- repo: https://github.com/pre-commit/pre-commit-hooks
+  rev: v4.2.0
+  hooks:
+  - id: check-yaml
+  - id: end-of-file-fixer
+  - id: trailing-whitespace
+- repo: https://github.com/PyCQA/isort
+  rev: 6.0.1
+  hooks:
+  - id: isort
+    args: ["--profile", "black"]
+- repo: https://github.com/psf/black
+  rev: 25.1.0
+  hooks:
+  - id: black
+    language_version: python3.11
+- repo: https://github.com/PyCQA/flake8
+  rev: 7.2.0
+  hooks:
+  - id: flake8

amqp/adapter/backpressure_handler.py

@@ -4,6 +4,7 @@ import time
 from asyncio import AbstractEventLoop
 from threading import Thread
 
+import psutil
 from aio_pika import Message
 from aio_pika.abc import AbstractRobustChannel
 
@@ -46,6 +47,40 @@ class ScaleRequestV1:
         )
 
 
+class RunningAverage:
+    def __init__(self, num_items):
+        self.buffer = [0] * num_items  # Fixed-size array
+        self.pointer = 0  # Current position in array
+        self.num_items = num_items  # Maximum number of items to store
+        self.is_filled = False  # Track if buffer is fully filled
+
+    def add(self, value):
+        # Store the value at current pointer position
+        self.buffer[self.pointer] = value
+
+        # Move pointer to next position with wrap-around
+        self.pointer += 1
+        if self.pointer >= self.num_items:
+            self.pointer = 0
+            self.is_filled = True
+
+    def get_average(self):
+        # Determine how many items we should average
+        count = self.num_items if self.is_filled else self.pointer
+
+        if count == 0:
+            return 0.0  # Avoid division by zero
+
+        return sum(self.buffer) / count
+
+    def get_current_values(self):
+        """Returns the values in chronological order (oldest first)"""
+        if not self.is_filled:
+            return self.buffer[: self.pointer]
+
+        return self.buffer[self.pointer :] + self.buffer[: self.pointer]
+
+
 class BackpressureHandler:
     # Track the number of messages currently being processed
     current_parallel_executions = 0
@@ -69,6 +104,7 @@ class BackpressureHandler:
         self.swarm_service_id = self.config.amq_adapter.swarm_service_id
         self.swarm_task_id = self.config.amq_adapter.swarm_task_id
         self.parallel_workers = self.config.backpressure.threshold_threads
+        self.average_cpu_usage = None
 
     def increase_parallel_executions(self):
         """Increase the number of parallel executions"""
@@ -119,67 +155,90 @@ class BackpressureHandler:
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
     def backpressure_monitor_loop(self):
-        _time_window_sec = self.config.backpressure.time_window
-        _idle_duration_millis = 1000 * self.config.backpressure.idle_duration
+        _time_window_millis = self.config.backpressure.time_window
+        _idle_duration_millis = self.config.backpressure.idle_duration
+        _overload_duration_millis = self.config.backpressure.idle_duration
         _loop = asyncio.new_event_loop()
+        _monitor_interval = 0.2  # Monitor every 200ms
+        CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2
+        IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90
+        self.average_cpu_usage = RunningAverage(
+            int(max(_overload_duration_millis, _idle_duration_millis) // _monitor_interval)
+        )
 
         async def _backpressure_monitor():
             """Monitor the backpressure conditions and trigger events accordingly"""
+            _cpu_usage_state = CPU_ACTIVE
+            _cpu_usage_changed = time.time()
             while True:
-                logging_info(
-                    "Backpressure: Monitor loop, current=%s",
-                    self.current_parallel_executions,
-                )
-                logging_info(
-                    "Backpressure: Last data message time=%s, eventTime=%s",
-                    self.last_data_message_time,
-                    self.last_backpressure_event_time,
-                )
-                logging_info(
-                    "Backpressure: Last backpressure event=%s",
-                    self.last_backpressure_event,
-                )
+                _current_cpu_usage = psutil.cpu_percent(interval=None)
+                self.average_cpu_usage.add(_current_cpu_usage)
+                _average_cpu_usage = self.average_cpu_usage.get_average()
+                _now = time.time()
+                # logging_info(
+                #    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
+                #    _current_cpu_usage,
+                #    _average_cpu_usage,
+                #    _cpu_usage_state,
+                #    self.last_data_message_time,
+                #    self.last_backpressure_event,
+                #    self.last_backpressure_event_time,
+                # )
+                if _average_cpu_usage < IDLE_THRESHOLD:
+                    if _cpu_usage_state != CPU_IDLE:
+                        _cpu_usage_changed = _now
+                    _cpu_usage_state = CPU_IDLE
+                elif _average_cpu_usage > OVERLOAD_THRESHOLD:
+                    if _cpu_usage_state != CPU_OVERLOAD:
+                        _cpu_usage_changed = _now
+                    _cpu_usage_state = CPU_OVERLOAD
+                else:
+                    if _cpu_usage_state != CPU_ACTIVE:
+                        _cpu_usage_changed = _now
+                    _cpu_usage_state = CPU_ACTIVE
+
                 # Check if the current time exceeds the last backpressure event time
                 if (
-                    self.current_parallel_executions >= self.parallel_workers
-                    and time.time() - self.last_backpressure_event_time
-                    > _time_window_sec
-                    and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
+                    _cpu_usage_state == CPU_OVERLOAD
+                    and _now - _cpu_usage_changed > _overload_duration_millis
+                    and (
+                        _now - self.last_backpressure_event_time > _time_window_millis
+                        or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
+                    )
                 ):
                     # Trigger the overload event
                     await self.handle_backpressure_overload_event()
-                    self.last_backpressure_event_time = time.time()
+                    self.last_backpressure_event_time = _now
                     self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
                 elif (
-                    self.current_parallel_executions == 0
-                    and time.time() - self.last_data_message_time
-                    >= _idle_duration_millis
-                    and self.last_backpressure_event != ScalingRequestAlert.IDLE
+                    _cpu_usage_state == CPU_IDLE
+                    and _now - _cpu_usage_changed > _idle_duration_millis
+                    and _now - self.last_data_message_time > _idle_duration_millis
+                    and (
+                        _now - self.last_backpressure_event_time > _time_window_millis
+                        or self.last_backpressure_event != ScalingRequestAlert.IDLE
+                    )
                 ):
                     # Trigger the idle event
                     await self._handle_backpressure_idle_event()
                     self.last_backpressure_event = ScalingRequestAlert.IDLE
-                    self.last_backpressure_event_time = time.time()
+                    self.last_backpressure_event_time = _now
                 else:
                     # Trigger update event
-                    if (
-                        self.current_parallel_executions > 0
-                        and time.time() - self.last_backpressure_event_time
-                        > _time_window_sec
-                    ):
+                    if _now - self.last_backpressure_event_time > _time_window_millis:
                         # Trigger the update event
                         _scaling_request: ScaleRequestV1 = ScaleRequestV1(
                             self.swarm_service_id,
                             self.swarm_task_id,
-                            self.parallel_workers,
-                            self.parallel_workers - self.current_parallel_executions,
+                            100,
+                            _average_cpu_usage,
                             ScalingRequestAlert.UPDATE,
                         )
                         # Address the message to any adapter capable of supporting BACKPRESSURE request
                         await self.publish_backpressure_request(_scaling_request)
-                        self.last_backpressure_event_time = time.time()
+                        self.last_backpressure_event_time = _now
                         self.last_backpressure_event = ScalingRequestAlert.UPDATE
-                await asyncio.sleep(_time_window_sec)
+                await asyncio.sleep(_monitor_interval)
 
         _loop.run_until_complete(_backpressure_monitor())
 
@@ -221,15 +280,11 @@ class BackpressureHandler:
         if not self.exchange:
 
             async def _wrap_rabbit_mq_api_init(channel):
-                _exchange = await channel.get_exchange(
-                    name="cleverthis.clevermicro.management"
-                )
+                _exchange = await channel.get_exchange(name="cleverthis.clevermicro.management")
                 return _exchange
 
             self.exchange = await await_result(
-                asyncio.run_coroutine_threadsafe(
-                    _wrap_rabbit_mq_api_init(self.channel), self.loop
-                )
+                asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api_init(self.channel), self.loop)
             )
 
         if self.exchange:
@@ -257,6 +312,4 @@ class BackpressureHandler:
                     return True
                 return False
 
-            await await_future(
-                asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)
-            )
+            await await_future(asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop))

amqp/adapter/cleverthis_service_adapter.py

@@ -6,53 +6,36 @@ Methods here will invoke the actual CleverThis Service code, which are passed as
 import asyncio
 import inspect
 import json
+import pathlib
+import re
 from asyncio import AbstractEventLoop
-from typing import Dict, List, Callable, Coroutine, Any
+from collections import defaultdict
+from typing import Any, Callable, Coroutine, Dict, List
 
 from aio_pika.abc import (
+    AbstractExchange,
     AbstractRobustChannel,
     AbstractRobustConnection,
-    AbstractExchange,
 )
-from aiohttp import ClientSession, ClientResponse
-from dataclasses import dataclass
-from fastapi import HTTPException
+from aiohttp import ClientResponse, ClientSession
+from fastapi import HTTPException, UploadFile
+from fastapi.routing import APIRoute
+from fastapi.security import OAuth2PasswordRequestForm
+from pydantic import BaseModel
+from starlette.datastructures import Headers
 from starlette.responses import FileResponse
 
+from amqp.adapter import serializer
 from amqp.adapter.data_parser import AMQDataParser
 from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import DataMessage, DataResponse
+from amqp.model.model import AMQErrorMessage, AMQMessage, DataMessage, DataResponse
 from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
 from amqp.router.utils import await_result
 
 
-@dataclass
-class AMQMessage(DataMessage):
-    """
-    Expand the core DataMessage type to include the channel, which is required to create a dedicated queue
-    in case the response should initiate file download.
-    """
-
-    def __init__(self, data_message: DataMessage, connection: AbstractRobustConnection):
-        # Copy all fields from DataMessage
-        self.magic = data_message.magic
-        self.id = data_message.id
-        self.method = data_message.method
-        self.domain = data_message.domain
-        self.path = data_message.path
-        self.headers = data_message.headers
-        self.trace_info = data_message.trace_info
-        self.base64_body = data_message.base64_body
-        # Add the new field
-        self.connection = connection
-        self.extra_data = (
-            data_message.extra_data if hasattr(data_message, "extra_data") else {}
-        )
-
-
 async def _convert_file_response(
     obj: FileResponse,
     connection: AbstractRobustConnection,
@@ -61,15 +44,16 @@ async def _convert_file_response(
 ) -> str:
     """
     Converts a FileResponse object to a JSON string containing the filename and stream name.
-    The file content itself is transmitted through the named stream (which is implemented as temporary dedicated queue.
+    The file content itself is transmitted through the named stream (which is implemented as
+    temporary dedicated queue.
     """
 
     async def _wrap_amq_api_calls(
         connection: AbstractRobustConnection,
     ) -> tuple[AbstractRobustChannel, AbstractExchange, str]:
-        # Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with correct event loop
-        # This is needed because the RabbitMQ API is not thread-safe and needs to be called in the context of
-        # the event loop the first RabbitMQ connection was created with.
+        # Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with
+        # correct event loop. This is needed because the RabbitMQ API is not thread-safe and needs
+        # to be called in the context of event loop the first RabbitMQ connection was created with.
         _channel = await connection.channel()
         _exchange_name: str = AMQ_REPLY_TO_EXCHANGE
         _exchange = await _channel.get_exchange(name=_exchange_name, ensure=True)
@@ -84,9 +68,10 @@ async def _convert_file_response(
         logging_debug(f"Publishing to exchange: {_exchange_name}, q: {_queue_name}")
         return _channel, _exchange, _queue_name
 
-    # All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached to parent
-    # event loop, and need to execute at that event loop, not the current one
-    # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
+    # All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached
+    # to parent event loop, and need to execute at that event loop, not the current one
+    # allow coroutine to execute. calling _future.result() will block the entire thread
+    # and the coroutine will never execute
     _channel, _exchange, _routing_key = await await_result(
         asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop)
     )
@@ -95,7 +80,7 @@ async def _convert_file_response(
         _exchange, obj.path, _routing_key, loop
     )
     # as above, but luckily this time we don't need to wait for the result
-    _future = asyncio.run_coroutine_threadsafe(_channel.close(), loop)
+    asyncio.run_coroutine_threadsafe(_channel.close(), loop)
     return json.dumps(
         {
             "files": [
@@ -121,19 +106,22 @@ def is_likely_json(text: str) -> bool:
 # =============  CleverThisServiceAdapter =============
 class CleverThisServiceAdapter:
     """
-    A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the
-    appropriate CleverThisService API endpoint.
-    IMPORTANT: This class is not intended to be used directly. It should be subclassed for each CleverThis service
-    and override the methods to handle the specific API calls.
+    A base class for CleverThis service adapter that handles the incoming AMQP messages and routes
+    them to the appropriate CleverThisService API endpoint.
+    IMPORTANT: This class is not intended to be used directly. It should be subclassed for
+    each CleverThis service and override the methods to handle the specific API calls.
     """
 
     def __init__(
-        self, amq_configuration: AMQConfiguration, session: ClientSession = None
+        self,
+        amq_configuration: AMQConfiguration,
+        routes: List[APIRoute],
+        session: ClientSession = None,
     ):
         """
-        Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP session.
-        The HTTP session is for the case when the adapter uses loose coupling with the CleverThis service via
-        HTTP REST API.
+        Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP
+        session. The HTTP session is for the case when the adapter uses loose coupling with
+        the CleverThis service via HTTP REST API.
         """
         self.amq_configuration = amq_configuration
         self.session = session
@@ -143,6 +131,79 @@ class CleverThisServiceAdapter:
         self.require_authenticated_user = (
             self.amq_configuration.amq_adapter.require_authenticated_user
         )
+        self.routes = self.group_routes_by_method(api_routes=routes)
+
+    def get_endpoint(self, endpoint: Any) -> Any:
+        """
+        Optionally, Service can override this method to alter the endpoint mapping.
+        This is useful for example when the CleverThis service has a JSON wrapper
+        around the endpoint, like CleverSwarm, which might be the preferred endpoint to call.
+
+        :param endpoint: The original endpoint to be called.
+        :return: The actual endpoint to be called.
+        """
+        return endpoint
+
+    def clone_and_adapt_route(self, original_route: APIRoute) -> APIRoute:
+        """
+        Creates a clone of APIRoute. Checks if there's preferred endpoint with json serialization.
+        It also adjusts the path regex to ensure it matches the path in different API versions.
+        """
+
+        # Need to adapt also the path regex to ensure it matches the path in different API versions
+        # Ensures the leading '^' is followed by any match and there's no '/' before final '$'
+        _new_regex_string = "^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$"
+        _new_route: APIRoute = APIRoute(
+            path=original_route.path,
+            endpoint=self.get_endpoint(
+                original_route.endpoint
+            ),  # Service can alter the endpoint mapping here
+            response_model=original_route.response_model,
+            status_code=original_route.status_code,
+            tags=original_route.tags.copy() if original_route.tags else None,
+            dependencies=(
+                original_route.dependencies.copy() if original_route.dependencies else None
+            ),
+            summary=original_route.summary,
+            description=original_route.description,
+            response_description=original_route.response_description,
+            responses=(original_route.responses.copy() if original_route.responses else None),
+            deprecated=original_route.deprecated,
+            name=original_route.name,
+            methods=original_route.methods.copy() if original_route.methods else None,
+            operation_id=original_route.operation_id,
+            response_model_include=original_route.response_model_include,
+            response_model_exclude=original_route.response_model_exclude,
+            response_model_by_alias=original_route.response_model_by_alias,
+            response_model_exclude_unset=original_route.response_model_exclude_unset,
+            response_model_exclude_defaults=original_route.response_model_exclude_defaults,
+            response_model_exclude_none=original_route.response_model_exclude_none,
+            include_in_schema=original_route.include_in_schema,
+            response_class=original_route.response_class,
+            dependency_overrides_provider=original_route.dependency_overrides_provider,
+            callbacks=(original_route.callbacks.copy() if original_route.callbacks else None),
+            openapi_extra=(
+                original_route.openapi_extra.copy() if original_route.openapi_extra else None
+            ),
+            generate_unique_id_function=original_route.generate_unique_id_function,
+        )
+        _new_route.path_regex = re.compile(_new_regex_string)
+        return _new_route
+
+    def group_routes_by_method(self, api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]:
+        """
+        Groups API routes by their HTTP method. This is important to ensure that URLPath is matched
+        correctly, as different methods may share the same URLPath (e.g. GET, POST PUT, DELETE).
+        """
+        _method_to_routes = defaultdict(list)  # Automatically initializes new lists for new keys
+        for _route in api_routes:
+            if (
+                isinstance(_route, APIRoute) and _route.methods
+            ):  # Ensure methods exist (should always be true for valid routes)
+                for method in _route.methods:
+                    _method_to_routes[method].append(self.clone_and_adapt_route(_route))
+
+        return dict(_method_to_routes)
 
     def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
         """
@@ -157,9 +218,11 @@ class CleverThisServiceAdapter:
 
     async def get_current_user(self, token: str) -> object:
         """
-        To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
-        in format appropriate for the service.
+        To Be Overriden in subclass for given CleverThis Service, to return the current user based
+        on the token, in format appropriate for the service.
+
         :param token: The token to be used for authentication.
+
         :return: A user object or None if the user is not authenticated.
         """
         return None
@@ -173,11 +236,9 @@ class CleverThisServiceAdapter:
         _auth_user: Any | None = None
         _amq_response: Any | None = None
 
-        logging_debug(
-            f"on_message: require_authenticated_user={self.require_authenticated_user}"
-        )
+        logging_debug(f"on_message: require_authenticated_user={self.require_authenticated_user}")
         if self.require_authenticated_user:
-            _auth: str = message.headers.get("Authorization", "")
+            _auth: str = message.headers().get("Authorization", "")
             logging_debug(f"Auth: {_auth}")
             if isinstance(_auth, List):
                 _auth = _auth[0]
@@ -189,12 +250,8 @@ class CleverThisServiceAdapter:
                 except Exception as error:
                     logging_error(f"on_message: Error getting user: {error}")
 
-        if (
-            _auth_user
-            or not self.require_authenticated_user
-            or message.path.endswith("/login")
-        ):
-            method = message.method.lower()
+        if _auth_user or not self.require_authenticated_user or message.path().endswith("/login"):
+            method = message.method().lower()
             if method == "get":
                 _amq_response = await self.get(message, _auth_user)
             elif method == "put":
@@ -206,113 +263,296 @@ class CleverThisServiceAdapter:
             elif method == "delete":
                 _amq_response = await self.delete(message, _auth_user)
             else:
-                raise ValueError(f"Unexpected HTTP method: {message.method}")
+                raise ValueError(f"Unexpected HTTP method: {message.method()}")
         else:
-            _amq_response = DataResponseFactory.of(
-                message.id, 401, "text/plain", b"Unauthorized", message.trace_info
+            _amq_response = DataResponseFactory.of_error_message(
+                message.id(),
+                AMQErrorMessage(401, "Unauthorized", "Unauthorized"),
+                message.trace_info(),
             )
         logging_info(
-            f"ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}"
+            f"ALL DONE in on_message id:{message.id()}, resp code:{_amq_response.response_code()}"
         )
         return _amq_response
 
-    # ==================================================================================================
-    # =====================  C l e v e r T h i s    A P I    m e t h o d s  ============================
-    # ==================================================================================================
-
-    async def get(self, message: DataMessage, auth_user: Any) -> DataResponse:
+    async def _authorize_forward_request(
+        self, message: DataMessage, request_coroutine, auth_user: Any
+    ) -> DataResponse:
         """
-        Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        It will attempt to call the REST endpoint directly using HTTP protocol.
+        This method is used to authorize the request and forward it to the CleverThis Service API.
+        Used only for loose / 3rd party mode of coupling with the Service
+        """
+        # AMQDataParser.parse_request_body(message)
+        # TODO: implement authorization logic here - ensure the security related headers are set
+        # This depends on what kind of security is required by the CleverThis service.
+        # However at this time there is no loosely coupled Service.
+
+        return await request_coroutine
+
+    # ==============================================================================================
+    # =====================  C l e v e r T h i s    A P I    m e t h o d s  ========================
+    # ==============================================================================================
+
+    async def get(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+        """
+        Handles the GET requests for the CleverThis API. In tight coupling (self.session is None),
+        use path, path params and query string to determine the endpoint method to invoke.
+        In loose coupling (self.session is not None), forward the request as REST request using
+        the session, i.e. attempt to call the REST endpoint directly using HTTP protocol.
+
         :param message: message from the AMQP that contains the GET request
         :param auth_user: user authenticated by the API Gateway, as UserSchem
+
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
 
-        # unmatched path - try to invoke the service directly on this path
-        url = f"http://{self.service_host}:{self.service_port}{message.path}"
-        headers = {**message.headers, **message.trace_info}
+        if self.session is None:
+            return await self.handle_no_body_payload(message, auth_user, "GET")
+
+        # Forwards as REST - try to invoke the service directly on this path
+        url = f"http://{self.service_host}:{self.service_port}{message.path()}"
+        headers = {**message.headers(), **message.trace_info()}
         logging_info(
             f"SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}"
         )
-        if self.session:
-            _http_resp: ClientResponse = await self.session.get(url, headers=headers)
-            return DataResponseFactory.of(
-                message.id,
-                _http_resp.status,
-                _http_resp.content_type,
-                await _http_resp.read(),
-                message.trace_info,
-            )
+        _http_resp: ClientResponse = await self.session.get(url, headers=headers)
         return DataResponseFactory.of(
-            message.id,
-            404,
-            "text/plain",
-            str("Destination location does not exists").encode("utf-8"),
-            message.trace_info,
+            message.id(),
+            _http_resp.status,
+            _http_resp.content_type,
+            await _http_resp.read(),
+            message.trace_info(),
         )
 
-    async def put(self, message: DataMessage, auth_user: Any) -> DataResponse:
+    async def put(self, message: AMQMessage, auth_user: Any) -> DataResponse:
         """
-        Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        It will attempt to call the REST endpoint directly using HTTP protocol.
+        Handles the PUT requests for the CleverThis API. Uses path, path params, query string and
+        body payload to determine the endpoint method to invoke and its input arguments
+        In loose coupling (self.session is not None), attempt to call the REST endpoint directly
+        using HTTP protocol.
+
         :param message: message from the AMQP that contains the PUT request
         :param auth_user: user authenticated by the API Gateway, as UserSchem
+
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        return await self.handle_possible_form(
+        if self.session is None:
+            return await self.handle_body_payload(message, auth_user, "PUT")
+
+        # Forwards as REST - try to invoke the service directly on this path
+        return await self._authorize_forward_request(
             message,
             self.session.put(
-                f"http://{self.service_host}:{self.service_port}{message.path}",
-                data=message.body,
-                headers={**message.headers, **message.trace_info},
+                f"http://{self.service_host}:{self.service_port}{message.path()}",
+                data=message.body(),
+                headers={**message.headers(), **message.trace_info()},
             ),
             auth_user,
         )
 
-    async def post(self, message: DataMessage, auth_user: Any) -> DataResponse:
+    async def post(self, message: AMQMessage, auth_user: Any) -> DataResponse:
         """
-        Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        It will attempt to call the REST endpoint directly using HTTP protocol.
-        :param message: message from the AMQP that contains the POST request, including body and headers
+        Handles the POST requests for the CleverThis API. Uses path, path params, query string and
+        body payload to determine the endpoint method to invoke and its input arguments
+        In loose coupling (self.session is not None), attempt to call the REST endpoint directly
+        using HTTP protocol.
+
+        :param message: message from AMQP that contains the POST request, including body and headers
         :param auth_user: user authenticated by the API Gateway, as UserSchem
+
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        return await self.handle_possible_form(
+        if self.session is None:
+            return await self.handle_body_payload(message, auth_user, "POST")
+
+        return await self._authorize_forward_request(
             message,
             self.session.post(
-                f"http://{self.service_host}:{self.service_port}{message.path}",
-                data=message.body,
-                headers={**message.headers, **message.trace_info},
+                f"http://{self.service_host}:{self.service_port}{message.path()}",
+                data=message.body(),
+                headers={**message.headers(), **message.trace_info()},
             ),
             auth_user,
         )
 
-    async def handle_possible_form(
-        self, message: DataMessage, request_coroutine, auth_user: Any
+    async def head(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+        """
+        Handles the HEAD requests for the CleverThis API. In tight coupling (self.session is None),
+        use path, path params and query string to determine the endpoint method to invoke.
+        In loose coupling (self.session is not None), HEAD call is not supported.
+
+        :param message: message from AMQP containing the HEAD request, including body and headers
+        :param auth_user: user authenticated by the API Gateway, as UserSchema
+
+        :return: DataResponse to be sent back to the API Gateway via AMQP
+        """
+        if self.session is None:
+            return await self.handle_no_body_payload(message, auth_user, "HEAD")
+
+        raise NotImplementedError
+
+    async def delete(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+        """
+        Handles the DELETE requests for the CleverThis API.
+
+        :param message: message from AMQP that contains the DELETE request, including body & headers
+        :param auth_user: user authenticated by the API Gateway, as UserSchema
+
+        :return: DataResponse to be sent back to the API Gateway via AMQP
+        """
+        if self.session is None:
+            return await self.handle_no_body_payload(message, auth_user, "DELETE")
+
+        return await self._authorize_forward_request(
+            message,
+            self.session.post(
+                f"http://{self.service_host}:{self.service_port}{message.path()}",
+                data=message.body(),
+                headers={**message.headers(), **message.trace_info()},
+            ),
+            auth_user,
+        )
+
+    async def options(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+        """
+        Handles the OPTIONS requests. In tight coupling (self.session is None),
+        use path, path params and query string to determine the endpoint method to invoke.
+        In loose coupling (self.session is not None), forward the request as REST request using
+        the session, i.e. attempt to call the REST endpoint directly using HTTP protocol.
+
+        :param message: message from AMQP that contains the OPTIONS request, including headers
+        :param auth_user: user authenticated by the API Gateway, as UserSchema
+
+        :return: DataResponse to be sent back to the API Gateway via AMQP
+        """
+        if self.session is None:
+            return await self.handle_no_body_payload(message, auth_user, "OPTIONS")
+
+        raise NotImplementedError
+
+    async def patch(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+        """
+        Handles the PATCH requests for the CleverBRAG API.
+
+        :param message: message from AMQP that contains the PATCH request, including body & headers
+        :param auth_user: user authenticated by the API Gateway, as UserSchema
+
+        :return: DataResponse to be sent back to the API Gateway via AMQP
+        """
+        if self.session is None:
+            return await self.handle_no_body_payload(message, auth_user, "PATCH")
+
+        return await self._authorize_forward_request(
+            message,
+            self.session.post(
+                f"http://{self.service_host}:{self.service_port}{message.path()}",
+                data=message.body(),
+                headers={**message.headers(), **message.trace_info()},
+            ),
+            auth_user,
+        )
+
+    # ==============================================================================================
+    # =================  C l e v e r S w a r m   p a y l o a d   h a n d l e r s  ==================
+    # ==============================================================================================
+
+    async def handle_no_body_payload(
+        self, message: AMQMessage, auth_user: Any, method: str
     ) -> DataResponse:
-        _args = AMQDataParser.parse_request_body(message)
-        return await request_coroutine
-
-    async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
-        Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        It will attempt to call the REST endpoint directly using HTTP protocol.
-        :param message: message from the AMQP that contains the POST request, including body and headers
+        Handles the requests that do not have a body payload.
+        :param message: message from the AMQP that contains the request
         :param auth_user: user authenticated by the API Gateway, as UserSchem
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        raise NotImplementedError
+        _routes = self.routes.get(method, [])
+        _path, _args = AMQDataParser.parse_get_url(message.path())
+        for _route in _routes:
+            _match = _route.path_regex.match(_path)
+            if _match:
+                _args.update(_match.groupdict())
+                _args["authenticated_user"] = auth_user
+                # If the path matches, call the corresponding function
+                return await CleverThisServiceAdapter.call_cleverthis_api(
+                    message,
+                    _args,
+                    api_method=_route.endpoint,
+                    success_code=_route.status_code,
+                    loop=self.loop,
+                )
 
-    async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse:
+        return DataResponseFactory.of(
+            message.id(),
+            404,
+            "text/plain",
+            str(f"Destination location [{message.path()}] does not exists").encode("utf-8"),
+            message.trace_info(),
+        )
+
+    async def handle_body_payload(
+        self, message: AMQMessage, auth_user: Any, method: str
+    ) -> DataResponse:
         """
-        Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        It will attempt to call the REST endpoint directly using HTTP protocol.
-        :param message: message from the AMQP that contains the POST request, including body and headers
+        Handles the requests that have a body payload, including Multipart data.
+        :param message: message from the AMQP that contains the request
         :param auth_user: user authenticated by the API Gateway, as UserSchem
+        :param method: HTTP method (POST, PUT, etc.)
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        raise NotImplementedError
+        _form_data = AMQDataParser.parse_request_body(message)
+        _routes = self.routes.get(method, [])
+        _path, _query_args = AMQDataParser.parse_get_url(message.path())
+        for _route in _routes:
+            _match = _route.path_regex.match(_path)
+            if _match:
+                # If the path matches, call the corresponding function
+                _form_data.update(_match.groupdict())
+                _form_data.update(_query_args)
+                for _body_param in _route.dependant.body_params:
+                    if _body_param.name in _form_data:
+                        _file_data = _form_data[_body_param.name]
+                        actual_filepath = message.extra_data[_body_param.name]
+                        _form_data[_body_param.name] = (
+                            _body_param.type_.__name__ == "UploadFile"
+                            and UploadFile(
+                                file=pathlib.Path(actual_filepath).open("rb"),
+                                size=_file_data["size"],
+                                filename=_file_data["filename"],
+                                headers=Headers({"Content-Type": _file_data["mediaType"]}),
+                            )
+                            or _form_data[_body_param.name]
+                        )
+
+                _verified_args = {}
+                _required_args = getattr(_route.endpoint, "__annotations__", {})
+                for arg in _required_args:
+                    if arg in _form_data:
+                        _verified_args[arg] = _form_data[arg]
+                    else:
+                        # needed for /login  input
+                        if arg == "authenticated_user":
+                            _verified_args["authenticated_user"] = auth_user
+                        if _required_args[arg].__name__ == "OAuth2PasswordRequestForm":
+                            try:
+                                _verified_args[arg] = OAuth2PasswordRequestForm(**_form_data)
+                            except Exception as e:
+                                print(f"Error: {e}")
+
+                return await CleverThisServiceAdapter.call_cleverthis_api(
+                    message,
+                    _verified_args,
+                    api_method=_route.endpoint,
+                    success_code=_route.status_code if _route.status_code else 200,
+                    loop=self.loop,
+                )
+
+        return DataResponseFactory.of(
+            message.id(),
+            404,
+            "text/plain",
+            str(f"Destination location [{message.path()}] does not exists").encode("utf-8"),
+            message.trace_info(),
+        )
 
     @staticmethod
     async def call_cleverthis_api(
@@ -323,13 +563,15 @@ class CleverThisServiceAdapter:
         loop: AbstractEventLoop | None = None,
     ) -> DataResponse:
         """
-        Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary.
-        Invokes the provided API method with the given arguments and authenticated user,
-        and returns the response as DataResponse.
+        Handles POST and PUT requests, which typically contain possibly complex payload data,
+        passed as a dictionary. Invokes the provided API method with the given arguments and
+        authenticated user, and returns the response as DataResponse.
+
         :param message: DataMessage containing the details of the call and the payload
         :param args: map of the arguments to pass to the API method
         :param api_method: Callable - the method to invoke. Last parameter is the authenticated user
         :param success_code: HTTP code to be returned when operation suceeds
+
         :return: DataResponse class
         """
         try:
@@ -345,21 +587,19 @@ class CleverThisServiceAdapter:
                     elif arg.annotation == float:
                         _verified_args[arg.name] = float(args[arg.name])
                     elif arg.annotation == bool:
-                        _verified_args[arg.name] = bool(args[arg.name])
+                        _verified_args[arg.name] = serializer.str_to_bool(args[arg.name])
                     # TODO: more?
                     else:
                         _verified_args[arg.name] = args[arg.name]
                 else:
-                    logging_info(
-                        f"{api_method.__name__}: Missing required argument: {arg.name}"
-                    )
+                    logging_info(f"{api_method.__name__}: Missing required argument: {arg.name}")
             logging_info(
-                f"INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}"
+                "INVOKE ENDPOINT: {%s}, ARGS: {%s}",
+                api_method.__name__,
+                ", ".join(f"{k}={v}" for k, v in _verified_args.items()),
             )
             _result: Any = await api_method(**_verified_args)
-            logging_info(
-                f"ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}"
-            )
+            logging_info(f"ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}")
 
             if isinstance(_result, FileResponse):
                 _json_result, _content_type = (
@@ -371,6 +611,9 @@ class CleverThisServiceAdapter:
                     ),
                     "application/octet-stream",
                 )
+            elif isinstance(_result, BaseModel):
+                _json_result = _result.json()
+                _content_type = "application/json"
             else:
                 _json_result = _result
                 _content_type = "application/json"
@@ -381,44 +624,56 @@ class CleverThisServiceAdapter:
                 else (
                     _json_result.body
                     if hasattr(_json_result, "body")
-                    else (
-                        _json_result.read()
-                        if hasattr(_json_result, "read")
-                        else _json_result
-                    )
+                    else (_json_result.read() if hasattr(_json_result, "read") else _json_result)
                 )
             )
             logging_info(
-                f"ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}"
+                f"ALL DONE in call_cleverthis_api id:{message.id()}, resp code:{success_code}"
             )
             return DataResponseFactory.of(
-                message.id,
+                message.id(),
                 success_code,
                 _content_type,
                 _binary_result,
-                message.trace_info,
+                message.trace_info(),
             )
         except HTTPException as http_error:
+            logging_error(f"Service REST endpoint invocation failed: {http_error}")
             _content = str(http_error.detail)
             if not is_likely_json(_content):
-                _content = '{"error": "' + _content + '"}'
-            logging_error(f"Service REST endpoint invocation failed: {http_error}")
+                return DataResponseFactory.of_error_message(
+                    message.id(),
+                    AMQErrorMessage(
+                        http_error.status_code,
+                        "Service Invocation Failed",
+                        _content,
+                    ),
+                    message.trace_info(),
+                )
             return DataResponseFactory.of(
-                message.id,
+                message.id(),
                 http_error.status_code,
                 "application/json",
                 _content.encode("utf-8"),
-                message.trace_info,
+                message.trace_info(),
             )
         except Exception as error:
             logging_error(f"Service endpoint invocation failed: {error}")
             _content = str(error)
             if not is_likely_json(_content):
-                _content = '{"error": "' + _content + '"}'
+                return DataResponseFactory.of_error_message(
+                    message.id(),
+                    AMQErrorMessage(
+                        500,
+                        "Service Invocation Failed",
+                        _content,
+                    ),
+                    message.trace_info(),
+                )
             return DataResponseFactory.of(
-                message.id,
+                message.id(),
                 500,
                 "application/json",
                 _content.encode("utf-8"),
-                message.trace_info,
+                message.trace_info(),
             )

amqp/adapter/data_message_factory.py

@@ -5,21 +5,21 @@ methods to create, serialize and deserialize DataMessage objects.
 import base64
 import json
 import re
-from io import BytesIO
 from datetime import datetime, timezone
+from io import BytesIO
 from typing import Dict, List
 
 import amqp.adapter.file_handler
-from amqp.model.model import DataMessage
-from amqp.model.snowflake_id import SnowflakeId
+from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
 from amqp.adapter.serializer import (
-    map_with_list_as_string,
     map_as_string,
+    map_with_list_as_string,
     parse_map_list_string,
     parse_map_string,
 )
+from amqp.model.model import DataMessage, DataMessageMagicByte
+from amqp.model.snowflake_id import SnowflakeId
 from amqp.router.utils import sanitize_as_rabbitmq_name
-from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
 
 
 class DataMessageFactory:
@@ -30,7 +30,6 @@ class DataMessageFactory:
 
     MACHINE_ID_MASK = (1 << MACHINE_ID_BITS) - 1
     SEQUENCE_MASK = (1 << SEQUENCE_BITS) - 1
-    MAGIC_BYTE_HTTP_REQUEST = "A"
 
     _instance = None
     snowflake_epoch = datetime(2024, 10, 1, 0, 0, tzinfo=timezone.utc).timestamp()
@@ -100,12 +99,10 @@ class DataMessageFactory:
         :return: DataMessage object
         """
         serializable_headers = headers.copy() if headers else {}
-        payload = base64.b64encode(
-            message.encode("utf-8") if isinstance(message, str) else message
-        )
+        payload = base64.b64encode(message.encode("utf-8") if isinstance(message, str) else message)
         trace_info = {}
         return DataMessage(
-            self.MAGIC_BYTE_HTTP_REQUEST,
+            DataMessageMagicByte.HTTP_REQUEST.value,
             self.generate_snowflake_id(),
             method,
             domain,
@@ -122,7 +119,7 @@ class DataMessageFactory:
         :param message: DataMessage input
         :return: routing key (as string) compliant to RabbitMQ allowed character set.
         """
-        return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
+        return sanitize_as_rabbitmq_name(f"{message.domain()}.{message.path()}")
 
     @staticmethod
     def get_timestamp_from_id(_id: SnowflakeId) -> int:
@@ -132,9 +129,7 @@ class DataMessageFactory:
         :return: the message timestamp
         """
         bits = _id.get_bits()
-        low = bits[0] >> (
-            DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS
-        )
+        low = bits[0] >> (DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS)
         high = bits[1] << (
             64 - DataMessageFactory.SEQUENCE_BITS - DataMessageFactory.MACHINE_ID_BITS
         )
@@ -149,13 +144,13 @@ class DataMessageFactory:
         """
         return (
             f"DataMessage{{"
-            f"id={message.id}, "
-            f"method='{message.method}', "
-            f"domain='{message.domain}', "
-            f"path='{message.path}', "
-            f"headers={{{map_with_list_as_string(message.headers)}}}, "
-            f"traceInfo={{{map_as_string(message.trace_info)}}}, "
-            f"base64body={message.base64_body.decode()}"
+            f"id={message.id()}, "
+            f"method='{message.method()}', "
+            f"domain='{message.domain()}', "
+            f"path='{message.path()}', "
+            f"headers={{{map_with_list_as_string(message.headers())}}}, "
+            f"traceInfo={{{map_as_string(message.trace_info())}}}, "
+            f"base64body={message.base64body().decode()}"
             f"}}"
         )
 
@@ -166,19 +161,19 @@ class DataMessageFactory:
         :param message: message to serialize
         :return: messages as bytes
         """
-        id_bytes = str(message.id).encode()
+        id_bytes = str(message.id()).encode()
         prolog = (
-            f"|m={message.method}|d={message.domain}|p={message.path}|h={{{map_with_list_as_string(message.headers)}}}|t={{{map_as_string(message.trace_info)}}}|b="
+            f"|m={message.method()}|d={message.domain()}|p={message.path()}|h={{{map_with_list_as_string(message.headers())}}}|t={{{map_as_string(message.trace_info())}}}|b="
         ).encode("utf-8")
         header_length = 1 + len(id_bytes) + len(prolog)
         prolog_len = header_length.to_bytes(2, byteorder="big")
-        msg_length = 2 + header_length + len(message.base64_body)
+        msg_length = 2 + header_length + len(message.base64body())
         with BytesIO(bytearray(msg_length)) as baos:
             baos.write(prolog_len)
-            baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8"))
+            baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
             baos.write(id_bytes)
             baos.write(prolog)
-            baos.write(message.base64_body)
+            baos.write(message.base64body())
             return baos.getvalue()
 
     @staticmethod
@@ -192,7 +187,7 @@ class DataMessageFactory:
         metadata = input_bytes[2 : prolog_len + 2].decode()
 
         pattern = re.compile(
-            f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
+            f"{DataMessageMagicByte.HTTP_REQUEST.value}"
             r"([0-9A-Fa-f]+)\|"
             r"m=([^|]*)\|"
             r"d=([^|]*)\|"
@@ -220,7 +215,7 @@ class DataMessageFactory:
         trace_info = parse_map_string(trace_info_str)
 
         return DataMessage(
-            DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
+            DataMessageMagicByte.HTTP_REQUEST.value,
             SnowflakeId.from_hex(id_str),
             method,
             domain,

amqp/adapter/data_parser.py

@@ -1,15 +1,15 @@
 import io
 import json
-import python_multipart
-
-from fastapi import UploadFile
-from python_multipart.multipart import Field, File
-from typing import Tuple, Dict
+from typing import Dict, Tuple
 from urllib.parse import parse_qs, urlparse
 from xml.etree import ElementTree
 
-from amqp.adapter.logging_utils import logging_error, logging_debug
-from amqp.model.model import DataMessage
+import python_multipart
+from fastapi import UploadFile
+from python_multipart.multipart import Field, File
+
+from amqp.adapter.logging_utils import logging_debug, logging_error
+from amqp.model.model import AMQMessage
 
 
 class MultipartFormDataParser:
@@ -22,15 +22,14 @@ class MultipartFormDataParser:
     included, the message is transmitted in the unchanged multipart/form-data format.
     """
 
-    def __init__(self, message: DataMessage):
+    def __init__(self, message: AMQMessage):
         self.form_data: dict = {}
         self.is_multipart = False
         self.is_json = False
         self.is_valid = True
         # Convert each list of strings to a single string joined by newline, then to bytes
         _headers = {
-            key: "\n".join(value).encode("utf-8")
-            for key, value in message.headers.items()
+            key: "\n".join(value).encode("utf-8") for key, value in message.headers().items()
         }
         try:
             if (
@@ -52,9 +51,7 @@ class MultipartFormDataParser:
                 )
                 self.is_multipart = True
         except Exception as e:
-            logging_error(
-                f"Error parsing multipart/form-data: {e}. Trying parsing as JSON."
-            )
+            logging_error(f"Error parsing multipart/form-data: {e}. Trying parsing as JSON.")
             try:
                 self.form_data = json.loads(message.body())
                 self.is_json = True
@@ -70,8 +67,9 @@ class MultipartFormDataParser:
 
     def on_file(self, file: File):
         logging_debug(str(file))
+        file.file_object.seek(0)
         self.form_data[file.field_name.decode("utf-8")] = UploadFile(
-            file.file_object, size=file.size, filename=file.file_name
+            file.file_object, size=file.size, filename=file.file_name.decode("utf-8")
         )
 
 
@@ -130,13 +128,11 @@ class AMQDataParser:
         _parsed = urlparse(url)
         _query_params = parse_qs(_parsed.query)
         # Convert values from lists to single values when there's only one value
-        _simplified_params = {
-            k: v[0] if len(v) == 1 else v for k, v in _query_params.items()
-        }
+        _simplified_params = {k: v[0] if len(v) == 1 else v for k, v in _query_params.items()}
         return _parsed.path, _simplified_params
 
     @staticmethod
-    def parse_request_body(message: DataMessage):
+    def parse_request_body(message: AMQMessage):
         """
         Parses the HTTP POST request body based on the MIME type.
         This is typically called fom the concrete CleverThis Service to extract parameters from the message body.
@@ -146,7 +142,7 @@ class AMQDataParser:
         Returns:
             dict: Parsed data as a dictionary.
         """
-        _content_type = message.headers.get("Content-Type", "")
+        _content_type = message.headers().get("Content-Type", "")
         if not _content_type:
             raise ValueError("Content-Type header is required")
         content_type = _content_type[0]
@@ -162,15 +158,12 @@ class AMQDataParser:
         elif content_type == "application/x-www-form-urlencoded":
             try:
                 _form_data: dict = {
-                    k: v[0] if len(v) == 1 else v
-                    for k, v in parse_qs(message.body()).items()
+                    k: v[0] if len(v) == 1 else v for k, v in parse_qs(message.body()).items()
                 }
                 if len(_form_data) == 0 and len(message.body()) > 0:
                     _form_data = json.loads(message.body())
                     if _form_data["files"] and _form_data["form"]:
-                        _form_data = MultipartFormDataParser.process_form_data(
-                            _form_data
-                        )
+                        _form_data = AMQDataParser.process_form_data(_form_data)
                 return _form_data
             except Exception as e:
                 logging_error(f"Invalid URL-encoded form data: {e}")

amqp/adapter/data_response_factory.py

@@ -5,13 +5,13 @@ method to create, serialize and deserialize DataResponse message
 import base64
 import re
 from io import BytesIO
-from amqp.model.model import DataResponse
+
+from amqp.adapter.serializer import map_as_string, parse_map_string
+from amqp.model.model import AMQErrorMessage, DataMessageMagicByte, DataResponse
 from amqp.model.snowflake_id import SnowflakeId
-from amqp.adapter.serializer import parse_map_string, map_as_string
 
 
 class DataResponseFactory:
-    MAGIC_BYTE_HTTP_REQUEST = "A"
 
     @staticmethod
     def create_async_response_message(_id: SnowflakeId) -> DataResponse:
@@ -20,8 +20,14 @@ class DataResponseFactory:
         :param _id: ID
         :return: Response message
         """
-        return DataResponse(
-            str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {}
+        return DataResponseFactory.of(
+            id=_id,
+            response_code=200,
+            content_type="application/text",
+            body="OK".encode("utf-8"),
+            trace_info={},
+            error="",
+            error_cause="",
         )
 
     @staticmethod
@@ -31,6 +37,8 @@ class DataResponseFactory:
         content_type: str,
         body: bytes,
         trace_info: dict[str, str],
+        error: str | None = None,
+        error_cause: str | None = None,
     ) -> DataResponse:
         """
         Create the DataResponse message supplying all key values
@@ -42,13 +50,31 @@ class DataResponseFactory:
         :return: DataResponseMessage object
         """
         return DataResponse(
-            id=id.as_string(),
-            response_code=response_code,
+            magic=DataMessageMagicByte.HTTP_REQUEST.value,
+            id=id,
+            response_code=str(response_code),
             content_type=content_type,
-            response=body,
-            error=None,
-            error_cause=None,
+            error=error,
+            error_cause=error_cause,
+            headers={},
             trace_info=trace_info,
+            base64body=base64.b64encode(body),
+        )
+
+    @staticmethod
+    def of_error_message(
+        id: SnowflakeId,
+        error_message: AMQErrorMessage,
+        trace_info: dict[str, str],
+    ) -> DataResponse:
+        return DataResponseFactory.of(
+            id=id,
+            response_code=error_message.response_code,
+            content_type=error_message.content_type,
+            body=error_message.serialize(),
+            trace_info=trace_info,
+            error=error_message.error,
+            error_cause=error_message.detail,
         )
 
     @staticmethod
@@ -58,19 +84,19 @@ class DataResponseFactory:
         :param response: Object to serialize
         :return: byte array as serialized message
         """
-        id_bytes = response.id.encode("utf-8")
+        id_bytes = response.id().as_string().encode("utf-8")
         prolog = (
-            f"|r={response.response_code}|c={response.content_type}|e={response.error}|f={response.error_cause}|t={{{map_as_string(response.trace_info)}}}|b="
+            f"|r={response.response_code()}|c={response.content_type()}|e={response.error()}|f={response.error_cause()}|t={{{map_as_string(response.trace_info())}}}|b="
         ).encode("utf-8")
         header_length = 1 + len(id_bytes) + len(prolog)
         prolog_len = header_length.to_bytes(2, byteorder="big")
-        msg_length = 2 + header_length + len(response.response)
+        msg_length = 2 + header_length + len(response.base64body())
         with BytesIO(bytearray(msg_length)) as baos:
             baos.write(prolog_len)
-            baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8"))
+            baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
             baos.write(id_bytes)
             baos.write(prolog)
-            baos.write(base64.b64encode(response.response))
+            baos.write(response.base64body())
             return baos.getvalue()
 
     @staticmethod
@@ -84,7 +110,7 @@ class DataResponseFactory:
         metadata = input_bytes[2 : prolog_len + 2].decode()
 
         pattern = re.compile(
-            f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
+            f"{DataMessageMagicByte.HTTP_REQUEST.value}"
             r"([0-9A-Fa-f]+)\|"
             r"r=([^|]*)\|"
             r"c=([^|]*)\|"
@@ -110,14 +136,14 @@ class DataResponseFactory:
 
         trace_info = parse_map_string(trace_info_str)
 
-        return DataResponse(
+        return DataResponseFactory.of(
             id_str,
             response_code,
             content_type,
             base64.b64decode(body_b64),
+            trace_info,
             error,
             error_cause,
-            trace_info,
         )
 
 

amqp/adapter/file_handler.py

@@ -4,22 +4,22 @@ implements the file downloader for the amqp adapter, where file is sent in chunk
 
 import asyncio
 import json
-import os
 import logging
-from asyncio import Future, AbstractEventLoop
+import os
+from asyncio import AbstractEventLoop
 from collections import defaultdict
 from pathlib import Path
 
-from aio_pika import connect_robust, Message
+from aio_pika import Message, connect_robust
 from aio_pika.abc import (
-    AbstractIncomingMessage,
-    AbstractRobustChannel,
-    AbstractQueue,
     AbstractExchange,
+    AbstractIncomingMessage,
+    AbstractQueue,
+    AbstractRobustChannel,
     DeliveryMode,
 )
 
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
 from amqp.router.utils import await_result
 
 # Global dictionary to track completed downloads
@@ -166,7 +166,6 @@ class StreamingFileHandler(FileHandler):
             message_data (dict): The entire message
             output_dir (str): The directory where the file should be saved.
         """
-        global download_locations
         filename = file_info["filename"]
         size = file_info["size"]
         stream_name = file_info["streamName"]
@@ -192,9 +191,7 @@ class StreamingFileHandler(FileHandler):
                 _eof = IN_PROGRESS
                 channel: AbstractRobustChannel = await connection.channel()
                 queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False)
-                logging_debug(
-                    f"Awaiting all file chunks being downloaded for {filename}"
-                )
+                logging_debug(f"Awaiting all file chunks being downloaded for {filename}")
                 async with queue.iterator() as queue_iter:
                     async for message in queue_iter:
                         async with message.process():
@@ -237,12 +234,8 @@ class StreamingFileHandler(FileHandler):
                 _res = bytearray()
                 try:
                     channel: AbstractRobustChannel = await _connection.channel()
-                    queue: AbstractQueue = await channel.get_queue(
-                        queue_name, ensure=False
-                    )
-                    logging_debug(
-                        f"Awaiting all buffer chunks being downloaded for {queue_name}"
-                    )
+                    queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False)
+                    logging_debug(f"Awaiting all buffer chunks being downloaded for {queue_name}")
                     async with queue.iterator() as queue_iter:
                         async for message in queue_iter:
                             async with message.process():
@@ -281,7 +274,6 @@ class StreamingFileHandler(FileHandler):
             loop: The asyncio event loop of the RabbitMQ API
             output_dir
         """
-        global download_locations
         try:
             _message_data = json.loads(message.body.decode())
             _amq_message_data = json.loads(json_data.decode())
@@ -298,22 +290,16 @@ class StreamingFileHandler(FileHandler):
                     files_to_download.append(files_info)
 
             if not files_to_download:
-                logging_info(
-                    " [%s] No files to download in this message.", message.delivery_tag
-                )
+                logging_info(" [%s] No files to download in this message.", message.delivery_tag)
                 # await message.ack()
                 return defaultdict()
 
             # Create a task for each file download
             _tasks = []
             for file_info in files_to_download:
-                logging_info(
-                    f" [{message.delivery_tag}] about to create task for {file_info}"
-                )
+                logging_info(f" [{message.delivery_tag}] about to create task for {file_info}")
                 _task = asyncio.create_task(
-                    self.download_file(
-                        loop, rabbitmq_url, file_info, _message_data, output_dir
-                    )
+                    self.download_file(loop, rabbitmq_url, file_info, _message_data, output_dir)
                 )
                 _tasks.append(_task)
             logging_info(
@@ -379,9 +365,7 @@ class StreamingFileHandler(FileHandler):
                         "total_size": _file_size,
                         "chunk_size": len(_chunk),
                         "eof": (
-                            END_OF_FILE
-                            if max_chunk_size + _offset >= _file_size
-                            else IN_PROGRESS
+                            END_OF_FILE if max_chunk_size + _offset >= _file_size else IN_PROGRESS
                         ),
                     },
                 )
@@ -394,9 +378,7 @@ class StreamingFileHandler(FileHandler):
                 while not _future.done():
                     # sleep allows coroutine to execute. must wait for completion to ensure all chunks written in right order
                     await asyncio.sleep(0.02)
-                    logging_debug(
-                        f"Waiting for chunk {_offset}/{_file_size} future= {_future}"
-                    )
+                    logging_debug(f"Waiting for chunk {_offset}/{_file_size} future= {_future}")
 
                 _offset += len(_chunk)
                 logging_debug(

amqp/adapter/logging_utils.py

@@ -4,8 +4,11 @@ import os
 import sys
 import threading
 import traceback
+from importlib.metadata import version
 
-verbose = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
+# Get version from installed package metadata
+__version__ = version("amq_adapter")  # Name must match pyproject.toml [project] name
+__verbose__ = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
 
 
 def get_context_info():
@@ -51,18 +54,18 @@ def logging_error(msg: str, *args) -> None:
     if _exec_info:
         _tb = traceback.extract_tb(sys.exc_info()[2])[-1]
         logging.error(
-            f"{_thread_id} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
+            f"{_thread_id}:{__version__} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
             *args,
         )
     else:
         module, line_number, function_name = get_context_info()
         if module and line_number and function_name:
             logging.error(
-                f"{_thread_id} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+                f"{_thread_id}:{__version__} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
                 *args,
             )
         else:
-            logging.error(f"{_thread_id} [E] {msg}", *args)
+            logging.error(f"{_thread_id}:{__version__} [E] {msg}", *args)
 
 
 def logging_warning(msg: str, *args) -> None:
@@ -73,17 +76,17 @@ def logging_warning(msg: str, *args) -> None:
         msg (str): The warning message to log.
     """
     _thread_id = threading.get_ident()
-    if not verbose:
-        logging.warning(f"{_thread_id} [W] {msg}", *args)
+    if not __verbose__:
+        logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.warning(
-            f"{_thread_id} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id}:{__version__} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.warning(f"{_thread_id} [W] {msg}", *args)
+        logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
 
 
 def logging_info(msg: str, *args) -> None:
@@ -94,17 +97,17 @@ def logging_info(msg: str, *args) -> None:
         msg (str): The info message to log.
     """
     _thread_id = threading.get_ident()
-    if not verbose:
-        logging.info(f"{_thread_id} [*] {msg}", *args)
+    if not __verbose__:
+        logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f"{_thread_id} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id}:{__version__} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.info(f"{_thread_id} [*] {msg}", *args)
+        logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
 
 
 def logging_debug(msg: str, *args) -> None:
@@ -118,8 +121,8 @@ def logging_debug(msg: str, *args) -> None:
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f"{_thread_id} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id}:{__version__} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.debug(f"{_thread_id} [DBG] {msg}", *args)
+        logging.debug(f"{_thread_id}:{__version__} [DBG] {msg}", *args)

amqp/adapter/pydantic_serializer.py

@@ -2,12 +2,13 @@
 This module provides serialization and deserialization functions for Pydantic models that extend BaseModel
 """
 
-import uuid
 import json
+import uuid
 from datetime import datetime
-from typing import Dict, Any
+from typing import Any, Dict
+from urllib.parse import parse_qs, urlparse
+
 from pydantic import BaseModel
-from urllib.parse import urlparse, parse_qs
 
 from amqp.adapter.logging_utils import logging_debug
 

amqp/adapter/serializer.py

@@ -4,7 +4,7 @@ Helper functions used to aid with (de)serialization of messages sent over AMQP.
 
 import struct
 from io import BytesIO
-from typing import List, Union, Dict
+from typing import Dict, List, Union
 
 
 def long_to_bytes(x: int) -> bytes:
@@ -63,3 +63,7 @@ def parse_map_list_string(input_str: str) -> Dict[str, List[str]]:
         for token in input_str[1:-1].split("],"):
             add_to_map(map_, token)
     return map_
+
+
+def str_to_bool(value: str) -> bool:
+    return value.lower() in ("true", "yes", "1", "t", "y", "on")

amqp/adapter/service_message_factory.py

@@ -1,13 +1,12 @@
 import base64
-import logging
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import logging_error, logging_info
+from amqp.adapter.serializer import map_as_string, parse_map_string
 from amqp.adapter.trace_info_adapter import TraceInfoAdapter
 from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
-from amqp.adapter.serializer import parse_map_string, map_as_string
 
 RS = "|"
 SPLIT_RS = r"\|"
@@ -99,9 +98,7 @@ class ServiceMessageFactory:
             i += 1
             reply_to = input_str_array[i] if i < len(input_str_array) else ""
             i += 1
-            trace_info = (
-                parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
-            )
+            trace_info = parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
             i += 1
             message = input_str_array[i] if i < len(input_str_array) else ""
             return ServiceMessage(
@@ -111,11 +108,7 @@ class ServiceMessageFactory:
                 recipient_name=recipient_name,
                 reply_to=reply_to,
                 trace_info=trace_info,
-                message=(
-                    base64.b64decode(message).decode()
-                    if payload_type & 0x80
-                    else message
-                ),
+                message=(base64.b64decode(message).decode() if payload_type & 0x80 else message),
             )
         except Exception as e:
             logging_error(
@@ -148,9 +141,7 @@ class ServiceMessageFactory:
         :return: formatted message type
         """
         num_type = (
-            message.message_type.value
-            if message.message_type
-            else ServiceMessageType.UNKNOWN.value
+            message.message_type.value if message.message_type else ServiceMessageType.UNKNOWN.value
         ) - 1
         fmt = (num_type & 0x7F) | (message.payload_type & 0x80)
         return f"{fmt:02X}"

amqp/config/amq_configuration.py

@@ -5,7 +5,7 @@ import os
 from amqp.adapter.logging_utils import logging_error, logging_warning
 
 """
-Convert configuration from properties file to an object. 
+Convert configuration from properties file to an object.
 """
 
 
@@ -41,9 +41,7 @@ class AMQAdapter:
             fallback=False,
         )
         # these 2 shall be provided by Docker swarm, format is a random alphanumeric string
-        self.swarm_service_id = os.getenv(
-            "SWARM_SERVICE_ID", default="clever-amqp-service"
-        )
+        self.swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service")
         self.swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter")
 
     def adapter_prefix(self) -> str:
@@ -68,15 +66,9 @@ class Dispatch:
 
         :param config: config parser to read configuration values
         """
-        self.use_dlq = config.getboolean(
-            "CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False
-        )
-        self.amq_host = config.get(
-            "CleverMicro-AMQ", self.PREFIX + "amq-host", fallback="rabbitmq"
-        )
-        self.amq_port = config.getint(
-            "CleverMicro-AMQ", self.PREFIX + "amq-port", fallback=5672
-        )
+        self.use_dlq = config.getboolean("CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False)
+        self.amq_host = config.get("CleverMicro-AMQ", self.PREFIX + "amq-host", fallback="rabbitmq")
+        self.amq_port = config.getint("CleverMicro-AMQ", self.PREFIX + "amq-port", fallback=5672)
         self.amq_port_tls = config.getint(
             "CleverMicro-AMQ", self.PREFIX + "amq-port-tls", fallback=5671
         )

amqp/config/application.properties

@@ -2,30 +2,41 @@
 # CleverMicro AMQ Adapter settings
 # ====================================================================
 [CleverMicro-AMQ]
-cm.amq-adapter.service-name=amq-adapter-python
-cm.amq-adapter.generator-id=164
-cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
+# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters
+
+# A name to identify this service for the AMQ Adapter
+cm.amq-adapter.service-name=cleverthis-service-name
+# The CleverMicro AMQ Adapter instance ID, used to identify this instance in the CleverMicro AMQ Adapter cluster
+cm.amq-adapter.generator-id=1
+# Route mapping. For route detail description see section 3.1 in Onboarding Guide for Python:
+# https://docs.cleverthis.com/en/architecture/microservices/onboarding/python
+# NOTE: this is a comma-separated list of route mappings (multiple, comma-separated routes are allowed here)
+cm.amq-adapter.route-mapping=
+# Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token
 cm.amq-adapter.require-authenticated-user=false
 
+# CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding
 cm.dispatch.use-dlq=true
-cm.dispatch.use-confirms=false
 cm.dispatch.amq-host=rabbitmq
 cm.dispatch.amq-port=5672
 cm.dispatch.amq-port-tls=5671
 cm.dispatch.download-dir=/tmp/downloaded_files
-# Not used for CleverSwarm nor CleverBRAG or in tight coupling mode.
+# Not used for CleveSwarm nor CleverBRAG or in tight coupling mode.
 # applicable only in loose coupling mode, specify the destination where to forward the REST request
-#cm.dispatch.service-host=cleverswarm.localhost
+#cm.dispatch.service-host=cleverthis-service-name.localhost
 #cm.dispatch.service-port=8080
-#
+
+# Backpressure monitor settings
 # Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert
 cm.backpressure.threshold=5
-# Define maximum CPU usage before triggering backpressure OVERLOAD alert
-cm.backpressure.threshold-cpu=90
+# Define maximum CPU usage (%) before triggering backpressure OVERLOAD alert
+cm.backpressure.threshold-cpu-overload=90
+# Define maximum CPU usage (%) before triggering backpressure IDLE alert
+cm.backpressure.threshold-cpu-idle=10
 # Define the time window between the Backpressure reports, in seconds
 cm.backpressure.time-window=10
 # Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert
 cm.backpressure.cpu-overload-duration=30
 # Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert
-cm.backpressure.idle-duration=30
+cm.backpressure.cpu-idle-duration=30
 

amqp/healthcheck.py

@@ -1,4 +1,5 @@
 import sys
+
 import requests
 
 url = "http://localhost:8080/amq-adapter-healthcheck"

amqp/model/model.py

@@ -1,8 +1,10 @@
 import base64
 import json
-from enum import Enum
-from typing import List, Dict
 from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, List
+
+from aio_pika.abc import AbstractRobustConnection
 
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
@@ -16,9 +18,29 @@ Define the structure of messages used in Clever Micro
 """
 
 
+class DataMessageMagicByte(Enum):
+    HTTP_REQUEST = "A"
+    RPC_REQUEST = "R"
+
+
+@dataclass
 class CleverMicroMessage:
     DEFAULT_CONTENT_TYPE = ["application/octet-stream"]
 
+    """
+    The common structure of a message that transfers the user request to the Service and back. All data related messages
+    conform to this base structure:
+
+    magic - identifies the message type (REST, RPC, DataResponse)
+    id - unique message ID
+    m_field - method name (for REST style calls) or service name (for RPC style calls) or response code for DataResponse
+    d_field - domain name (for REST style calls) or module name (for RPC style calls) or  error code for DataResponse
+    p_field - path (for REST style calls) or method name (for RPC style calls) or error cause for DataResponse
+    headers - headers of the message (e.g. Content-Type, Accept, etc.)
+    trace_info - trace information (e.g. trace ID, span ID, etc. for Jaeger trace monitor)
+    base64body - the body of the message encoded in base64
+    """
+
     def __init__(
         self,
         magic: str,
@@ -66,11 +88,19 @@ class CleverMicroMessage:
     def body(self) -> bytes:
         return base64.b64decode(self.base64body())
 
-    def contentType(self) -> str:
+    def content_type(self) -> str:
         return self.headers().get("Content-Type", self.DEFAULT_CONTENT_TYPE)[0]
 
 
 class DataMessage(CleverMicroMessage):
+    """
+    The class represents a structure of a message that transfers the user request to the Service.
+    It may also be used to make a call/request between services.
+    The response to this call is in format of DataResponse class.
+
+    See DataMessageFactory for code to instantiate, serialize and deserialize the class
+    """
+
     def __init__(
         self,
         magic: str,
@@ -82,9 +112,7 @@ class DataMessage(CleverMicroMessage):
         trace_info: Dict[str, str],
         base64body: bytes,
     ):
-        super().__init__(
-            magic, id, method, domain, path, headers, trace_info, base64body
-        )
+        super().__init__(magic, id, method, domain, path, headers, trace_info, base64body)
 
     def method(self) -> str:
         return self.m_field()
@@ -95,19 +123,77 @@ class DataMessage(CleverMicroMessage):
     def path(self) -> str:
         return self.p_field()
 
+    def routing_key(self) -> str:
+        return sanitize_as_rabbitmq_name(f"{self.d_field()}.{self.p_field()}")
+
+
+class MultipartDataMessage(DataMessage):
+    def __init__(self, message: DataMessage, extra_data: dict):
+        super().__init__(
+            message.magic(),
+            message.id(),
+            message.method(),
+            message.domain(),
+            message.path(),
+            message.headers(),
+            message.trace_info(),
+            message.base64body(),
+        )
+        self.extra_data = extra_data
+
+
+@dataclass
+class AMQMessage(DataMessage):
+    """
+    Expand the core DataMessage type to include the channel, which is required to create a dedicated queue
+    in case the response should initiate file download.
+    """
+
+    _connection: AbstractRobustConnection = field(repr=False)  # Don't show in repr
+
+    def __init__(self, data_message: DataMessage, connection: AbstractRobustConnection):
+        # Initialize the parent DataMessage with all properties from the input data_message
+        super().__init__(
+            magic=data_message.magic(),
+            id=data_message.id(),
+            method=data_message.method(),
+            domain=data_message.domain(),
+            path=data_message.path(),
+            headers=data_message.headers(),
+            trace_info=data_message.trace_info(),
+            base64body=data_message.base64body(),
+        )
+        self._connection = connection
+        self.extra_data = data_message.extra_data if hasattr(data_message, "extra_data") else {}
+
+    @property
+    def connection(self) -> AbstractRobustConnection:
+        """Get the AMQP connection associated with this message"""
+        return self._connection
+
 
 class DataResponse(CleverMicroMessage):
+    """
+    The response message to REST & RPC style calls (note: the request is made with DataMessage)
+
+    See DataResponseFactory for code to instantiate, serialize and deserialize the class
+    """
+
     def __init__(
         self,
         magic: str,
         id: SnowflakeId,
         response_code: str,
+        content_type: str,
         error: str,
         error_cause: str,
         headers: Dict[str, List[str]],
         trace_info: Dict[str, str],
         base64body: bytes,
     ):
+        if headers is None:
+            headers = {}
+        headers["Content-Type"] = [content_type]
         super().__init__(
             magic,
             id,
@@ -179,65 +265,6 @@ class AMQRoute:
         return self.__str__()
 
 
-@dataclass
-class DataMessage:
-    """
-    The class represents a structure of a message that transfers the user request to the Service.
-    It may also be used to make a call/request between services.
-    The response to this call is in format of DataResponse class.
-    Please see DataMessageFactory for code to instantiate, serialize and deserialize the class
-    """
-
-    magic: str
-    id: SnowflakeId
-    method: str
-    domain: str
-    path: str
-    headers: dict[str, List[str]]
-    trace_info: dict[str, str]
-    base64_body: bytes
-
-    def routing_key(self) -> str:
-        return sanitize_as_rabbitmq_name(f"{self.domain}.{self.path}")
-
-    def body(self) -> bytes:
-        return base64.b64decode(self.base64_body)
-
-
-class MultipartDataMessage(DataMessage):
-    def __init__(self, message: DataMessage, extra_data: dict):
-        super().__init__(
-            message.magic,
-            message.id,
-            message.method,
-            message.domain,
-            message.path,
-            message.headers,
-            message.trace_info,
-            message.base64_body,
-        )
-        self.extra_data = extra_data
-
-
-@dataclass
-class DataResponse:
-    """
-    The response message to RPC style call (note: the request is made with DataMessage)
-    Please see DataResponseFactory for code to instantiate, serialize and deserialize the class
-    """
-
-    id: str
-    response_code: int
-    content_type: str
-    response: bytes
-    error: str | None
-    error_cause: str | None
-    trace_info: dict[str, str]
-
-    def body(self) -> bytes:
-        return base64.b64decode(self.response)
-
-
 # Ensure backward compatibility
 AMQResponse = DataResponse
 
@@ -266,7 +293,7 @@ class ScalingRequestAlert(Enum):
     IDLE = "IDLE"
     UPDATE = "UPDATE"
     SCALE_UP = "SCALE_UP"
-    SCSLE_DOWN = "SCALE_DOWN"
+    SCALE_DOWN = "SCALE_DOWN"
 
 
 @dataclass(frozen=True)
@@ -301,7 +328,32 @@ class ScalingRequest:
             task_id=data["taskId"],
             max_availability=data["maxAvailability"],
             current_availability=data["currentAvailability"],
-            request_type=ScalingRequestAlert(
-                data["requestType"]
-            ),  # Convert string to Enum
+            request_type=ScalingRequestAlert(data["requestType"]),  # Convert string to Enum
         )
+
+
+class AMQErrorMessage:
+    """
+    The error response message to RPC style call (note: the request is made with DataMessage)
+    """
+
+    def __init__(
+        self,
+        response_code: int,
+        error: str,
+        detail: str,
+        content_type: str = "application/json",
+    ):
+        self.response_code = response_code
+        self.error = error
+        self.detail = detail
+        self.content_type = content_type
+
+    def serialize(self) -> bytes:
+        return json.dumps(
+            {
+                "response_code": self.response_code,
+                "error": self.error,
+                "detail": self.detail,
+            }
+        ).encode("utf-8")

amqp/rabbitmq/rabbit_mq_client.py

@@ -1,20 +1,21 @@
 import logging
 
 import aio_pika
-from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType
+from aio_pika.abc import AbstractRobustExchange, AbstractRobustQueue
+from pika.exchange_type import ExchangeType
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import (
-    logging_error,
     logging_debug,
+    logging_error,
     logging_info,
     logging_warning,
 )
-from amqp.model.model import DataMessage, AMQRoute
+from amqp.model.model import AMQRoute, DataMessage
 from amqp.router.router_base import (
     AMQ_REPLY_TO_EXCHANGE,
-    DLQ_EXCHANGE,
     AMQ_RPC_EXCHANGE,
+    DLQ_EXCHANGE,
 )
 from amqp.router.router_consumer import RouterConsumer
 from amqp.router.utils import sanitize_as_rabbitmq_name
@@ -54,7 +55,7 @@ class RabbitMQClient:
             name=cm_reply_to_queue, durable=True, exclusive=False, auto_delete=False
         )
         self.reply_to_exchange = await self.channel.declare_exchange(
-            AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.DIRECT
+            AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.direct
         )
         await self.reply_to_queue.bind(
             exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue
@@ -84,21 +85,15 @@ class RabbitMQClient:
         After this the Service is listening on (consuming from) the queue for incoming messages.
         """
         if self.router.is_open(self.channel):
-            queue_args = (
-                self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
-            )
-            logging_info(
-                "RabbitMQClient register routes, cfg mapping: [%s]", route_mapping
-            )
+            queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None
+            logging_info("RabbitMQClient register routes, cfg mapping: [%s]", route_mapping)
             # convert the comma-separated string into list of AMQRoute objects
             requested_routes = self.router.unwrap_route_list(route_mapping)
             for route in requested_routes:
                 try:
                     # Note: AMQRoute allows one of the exchange or queue to be None, in which case the default is used
                     queue_name = (
-                        route.queue
-                        if route.queue
-                        else self.router.get_consume_queue_name()
+                        route.queue if route.queue else self.router.get_consume_queue_name()
                     )
                     logging_info("RabbitMQClient register route: [%s]", route)
                     queue: AbstractRobustQueue = await self.channel.declare_queue(
@@ -108,13 +103,9 @@ class RabbitMQClient:
                         auto_delete=False,
                         arguments=queue_args,
                     )
-                    _exchange_name: str = (
-                        route.exchange if route.exchange else AMQ_RPC_EXCHANGE
-                    )
-                    exchange: AbstractRobustExchange = (
-                        await self.channel.declare_exchange(
-                            name=_exchange_name, type=ExchangeType.TOPIC
-                        )
+                    _exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE
+                    exchange: AbstractRobustExchange = await self.channel.declare_exchange(
+                        name=_exchange_name, type=ExchangeType.topic
                     )
                     await queue.bind(exchange, route.key)
                     _c_tag = await queue.consume(callback, no_ack=False)
@@ -199,16 +190,14 @@ class RabbitMQClient:
                             message
                         ),  # context path is a routing key
                     )
-                    logging_info(
-                        f"Msg Publish (no-reply), messageId: {message.id}, to: {route}"
-                    )
+                    logging_info(f"Msg Publish (no-reply), messageId: {message.id()}, to: {route}")
                 else:
                     # This is RPC call, there should be response
                     pika_message: aio_pika.message.Message = aio_pika.message.Message(
                         body=DataMessageFactory.serialize(message),
                         content_encoding="utf-8",
                         delivery_mode=2,
-                        correlation_id=str(message.id),
+                        correlation_id=str(message.id()),
                         reply_to=self.router.get_reply_to_queue_name(),
                     )
                     await self.rpc_exchange.publish(
@@ -219,12 +208,14 @@ class RabbitMQClient:
                     )
                     logging_info(
                         "Msg Publish (RPC call), messageId: %s, to: %s",
-                        message.id.as_string(),
+                        message.id().as_string(),
                         route.as_string(),
                     )
             else:
                 logging_warning(
-                    "Message not sent, no route for %s/%s", message.domain, message.path
+                    "Message not sent, no route for %s/%s",
+                    message.domain(),
+                    message.path(),
                 )
 
         except Exception as err:

amqp/router/route_database.py

@@ -1,4 +1,3 @@
-import logging
 import re
 from typing import Set
 
@@ -41,13 +40,9 @@ class RouteDatabase:
             return re.compile(".*")
         return re.compile(regex)
 
-    def matches(
-        self, routing_key_from_route: str, routing_key_from_message: str
-    ) -> bool:
+    def matches(self, routing_key_from_route: str, routing_key_from_message: str) -> bool:
         route_pattern = self.pattern(routing_key_from_route)
-        return bool(
-            route_pattern.match(sanitize_as_rabbitmq_name(routing_key_from_message))
-        )
+        return bool(route_pattern.match(sanitize_as_rabbitmq_name(routing_key_from_message)))
 
     def wrap_routes(self) -> str:
         return ",".join(str(route) for route in self.routes)

amqp/router/router_base.py

@@ -1,8 +1,7 @@
-import logging
 from typing import Callable, List, Set
 
-from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.amq_route_factory import NULL_ROUTE, AMQRouteFactory
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
 from amqp.model.model import AMQRoute, DataMessage
 from amqp.router.route_database import RouteDatabase
 from amqp.router.utils import sanitize_as_rabbitmq_name
@@ -28,7 +27,7 @@ class RouterBase:
         self.route_removed_notifier: Callable[[AMQRoute], None] | None = None
 
     def get_routing_key(self, message: DataMessage) -> str:
-        return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
+        return sanitize_as_rabbitmq_name(f"{message.domain()}.{message.path()}")
 
     def wrap_routes(self) -> str:
         return self.route_database.wrap_routes()
@@ -38,10 +37,7 @@ class RouterBase:
 
     def unwrap_route_list(self, route_string: str) -> List[AMQRoute]:
         logging_debug("RouterMaster.unwrapRoutes, route(s): %s", route_string)
-        routes = [
-            AMQRouteFactory.from_string(route_str)
-            for route_str in route_string.split(",")
-        ]
+        routes = [AMQRouteFactory.from_string(route_str) for route_str in route_string.split(",")]
         return [route for route in routes if route != NULL_ROUTE]
 
     def get_routes(self) -> Set[AMQRoute]:
@@ -51,14 +47,12 @@ class RouterBase:
         return self.route_database.find_route(domain, path)
 
     def find_route_by_message(self, message: DataMessage) -> AMQRoute | None:
-        return self.route_database.find_route(message.domain, message.path)
+        return self.route_database.find_route(message.domain(), message.path())
 
     def add_routes(self, message: str) -> None:
         try:
             amq_route_list = self.unwrap_route_list(message)
-            logging_info(
-                " [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message
-            )
+            logging_info(" [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message)
             for route in amq_route_list:
                 logging_info("RouterMaster.addRoute, adding route: %s", route)
                 if not route.exchange and not route.queue and not route.key:
@@ -82,14 +76,9 @@ class RouterBase:
 
     def remove_routes(self, message: str) -> None:
         try:
-            to_remove = [
-                AMQRouteFactory.from_string(route_str)
-                for route_str in message.split(",")
-            ]
+            to_remove = [AMQRouteFactory.from_string(route_str) for route_str in message.split(",")]
             removed_count = sum(1 for route in to_remove if self.remove_route(route))
-            logging_info(
-                f"RouterMaster.removeRoutes(): {message}, removed={removed_count}"
-            )
+            logging_info(f"RouterMaster.removeRoutes(): {message}, removed={removed_count}")
         except IOError as err:
             logging_error(f"Can't remove route(s): {err}")
 

amqp/router/router_consumer.py

@@ -1,26 +1,26 @@
-import logging
 from typing import Set
 
 from aio_pika.abc import (
+    AbstractIncomingMessage,
     AbstractRobustChannel,
     AbstractRobustQueue,
-    AbstractIncomingMessage,
 )
+
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.logging_utils import (
+    logging_debug,
     logging_error,
     logging_info,
-    logging_debug,
     logging_warning,
 )
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import ServiceMessage, AMQRoute
+from amqp.model.model import AMQRoute, ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import (
+    ANY_RECIPIENT,
     CM_C_REQ_QUEUE,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
-    ANY_RECIPIENT,
 )
 from amqp.router.router_producer import RouterProducer
 
@@ -77,17 +77,13 @@ class RouterConsumer(RouterProducer):
             except Exception as ioe:
                 logging_error("RouterConsumer FATAL: cannot initialize, %s", ioe)
         else:
-            logging_error(
-                "RouterConsumer FATAL: cannot initialize, RabbitMQ channel is not open."
-            )
+            logging_error("RouterConsumer FATAL: cannot initialize, RabbitMQ channel is not open.")
 
     """
      * ServiceMessageType.ROUTING_DATA_REQUEST  handler  listening on queue 'CleverMicroRequest'
     """
 
-    async def handle_routing_data_request_message(
-        self, message: AbstractIncomingMessage
-    ):
+    async def handle_routing_data_request_message(self, message: AbstractIncomingMessage):
         # some debug statements, TODO - remove for prod release
         delivery_tag = message.delivery_tag
         debug = "Delivery.Properties = " + message.properties.__str__()
@@ -101,9 +97,7 @@ class RouterConsumer(RouterProducer):
 
         await message.ack(False)
 
-        cm_message: ServiceMessage = self.service_message_factory.from_bytes(
-            message.body
-        )
+        cm_message: ServiceMessage = self.service_message_factory.from_bytes(message.body)
         route_mapping: str = self.wrap_own_routes()
         logging_debug(
             "******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]",
@@ -124,9 +118,7 @@ class RouterConsumer(RouterProducer):
                         route_mapping,
                         cm_message.reply_to,
                     )
-                    await self.publish_service_message(
-                        service_message, self.cm_response_exchange
-                    )
+                    await self.publish_service_message(service_message, self.cm_response_exchange)
                 except Exception as err:
                     logging_error(
                         "******[%s] (ROUTING_DATA_REQ) Response to queue: %s failed: %s",
@@ -158,9 +150,7 @@ class RouterConsumer(RouterProducer):
                 serviceMessage: ServiceMessage = self.service_message_factory.of(
                     ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT
                 )
-                await self.publish_service_message(
-                    serviceMessage, self.cm_response_exchange
-                )
+                await self.publish_service_message(serviceMessage, self.cm_response_exchange)
                 self.add_own_route(route)
                 logging_info("RouterConsumer registered Route: %s", route)
             else:
@@ -182,9 +172,7 @@ class RouterConsumer(RouterProducer):
             service_message: ServiceMessage = self.service_message_factory.of(
                 ServiceMessageType.ROUTING_DATA_REMOVE, route.as_string(), ANY_RECIPIENT
             )
-            if not await self.publish_service_message(
-                service_message, self.cm_response_exchange
-            ):
+            if not await self.publish_service_message(service_message, self.cm_response_exchange):
                 logging_warning(
                     "RouterConsumer couldn't remove current route: %s, channel not open.",
                     route,
@@ -201,13 +189,9 @@ class RouterConsumer(RouterProducer):
         * Cleanup - de-register routes with master.
         """
         routes: Set[AMQRoute] = self.get_routes()
-        logging_info(
-            "RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes()
-        )
+        logging_info("RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes())
         not_removed = sum(
-            1
-            for r in (self.remove_route_on_cleanup(route) for route in routes)
-            if not r
+            1 for r in (self.remove_route_on_cleanup(route) for route in routes) if not r
         )
         if not_removed > 0:
             logging_warning(

amqp/router/router_producer.py

@@ -7,21 +7,20 @@
 * forwards it to the application.
 """
 
-import logging
-
 from aio_pika import Message
 from aio_pika.abc import (
+    AbstractIncomingMessage,
     AbstractRobustChannel,
-    AbstractRobustQueue,
     AbstractRobustConnection,
     AbstractRobustExchange,
-    AbstractIncomingMessage, ExchangeType,
+    AbstractRobustQueue,
 )
+from pika.exchange_type import ExchangeType
 
 from amqp.adapter.logging_utils import (
+    logging_debug,
     logging_error,
     logging_info,
-    logging_debug,
     logging_warning,
 )
 from amqp.adapter.service_message_factory import ServiceMessageFactory
@@ -29,13 +28,13 @@ from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import (
-    RouterBase,
+    ANY_RECIPIENT,
+    CM_C_AMQ_QUEUE,
+    CM_P_REPLY_TO,
+    CM_P_RESP_QUEUE,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
-    CM_C_AMQ_QUEUE,
-    CM_P_RESP_QUEUE,
-    CM_P_REPLY_TO,
-    ANY_RECIPIENT,
+    RouterBase,
 )
 
 
@@ -48,16 +47,14 @@ class RouterProducer(RouterBase):
         """
         super().__init__()
         self.connection: AbstractRobustConnection | None = None
-        self.service_message_factory = ServiceMessageFactory(
-            configuration.amq_adapter.service_name
-        )
+        self.service_message_factory = ServiceMessageFactory(configuration.amq_adapter.service_name)
         self.amq_configuration: AMQConfiguration = configuration
         self.channel: AbstractRobustChannel | None = None
         self.cm_request_exchange: AbstractRobustExchange | None = None
         self.cm_response_exchange: AbstractRobustExchange | None = None
 
         logging_info(
-            f"RouterProducer.<init>: %s, route: %s",
+            "RouterProducer.<init>: %s, route: %s",
             self.amq_configuration.amq_adapter.service_name,
             self.amq_configuration.amq_adapter.route_mapping,
         )
@@ -81,13 +78,13 @@ class RouterProducer(RouterBase):
                 # **** set up the service discovery internal exchanges ****
                 # 1. declare RequestExchange to where to publish the RoutingData requests
                 self.cm_request_exchange = await self.channel.declare_exchange(
-                    name=CM_REQUEST_EXCHANGE, type=ExchangeType.FANOUT
+                    name=CM_REQUEST_EXCHANGE, type=ExchangeType.fanout
                 )
                 #
                 # 2. declare ResponseExchange where to publish current RoutingData (response to
                 #    a RoutingData request)
                 self.cm_response_exchange = await self.channel.declare_exchange(
-                    name=CM_RESPONSE_EXCHANGE, type=ExchangeType.FANOUT
+                    name=CM_RESPONSE_EXCHANGE, type=ExchangeType.fanout
                 )
                 # 2a. declare a Response queue for Routing Data replies from Services
                 _cm_response_queue: str = self.get_response_queue_name()
@@ -99,9 +96,7 @@ class RouterProducer(RouterBase):
                     arguments={},
                 )
                 # 2b. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key)
-                await _response_queue.bind(
-                    exchange=CM_RESPONSE_EXCHANGE, routing_key="#"
-                )
+                await _response_queue.bind(exchange=CM_RESPONSE_EXCHANGE, routing_key="#")
                 # 2c. listen for incoming RoutingData messages
                 _c_tag = await _response_queue.consume(
                     callback=self.handle_routing_data_message, no_ack=False
@@ -115,9 +110,7 @@ class RouterProducer(RouterBase):
                 logging_error("RouterProducer FATAL: cannot initialize, %s", ioe)
 
         else:
-            logging_error(
-                "RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open."
-            )
+            logging_error("RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.")
 
     async def handle_routing_data_message(self, message: AbstractIncomingMessage):
         """
@@ -125,15 +118,13 @@ class RouterProducer(RouterBase):
         * routing data, invokes addRoute or removeRoute depending on the type of the message.
         """
         logging_info(
-            f"[%s] Received ROUTING DATA: [%s], msgId=%s",
+            "[%s] Received ROUTING DATA: [%s], msgId=%s",
             message.delivery_tag,
             message.body,
             message.message_id,
         )
         await message.ack(multiple=False)
-        cm_message: ServiceMessage = self.service_message_factory.from_bytes(
-            message.body
-        )
+        cm_message: ServiceMessage = self.service_message_factory.from_bytes(message.body)
         # ignore the message if it comes from ourselves (even if from different instance)
         logging_debug(
             "******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s",
@@ -144,9 +135,8 @@ class RouterProducer(RouterBase):
             cm_message.reply_to,
         )
         if (
-                cm_message.is_valid()
-                and not self.amq_configuration.amq_adapter.service_name
-                        == cm_message.reply_to
+            cm_message.is_valid()
+            and not self.amq_configuration.amq_adapter.service_name == cm_message.reply_to
         ):
             if cm_message.message_type == ServiceMessageType.ROUTING_DATA:
                 self.add_routes(cm_message.message)
@@ -165,9 +155,7 @@ class RouterProducer(RouterBase):
             serviceMessage: ServiceMessage = self.service_message_factory.of(
                 ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT
             )
-            if not await self.publish_service_message(
-                    serviceMessage, self.cm_request_exchange
-            ):
+            if not await self.publish_service_message(serviceMessage, self.cm_request_exchange):
                 logging_warning(
                     "RouterProducer could not request route data: %s channel is not open.",
                     CM_REQUEST_EXCHANGE,
@@ -177,7 +165,7 @@ class RouterProducer(RouterBase):
             logging_error("RouterProducer failed request routing data: %s", ioe)
 
     async def publish_service_message(
-            self, message: ServiceMessage, exchange: AbstractRobustExchange
+        self, message: ServiceMessage, exchange: AbstractRobustExchange
     ) -> bool:
         """
         * Publishes a service message to the service queue and logs success log entry.

amqp/service/amq_message_handler.py

@@ -1,32 +1,32 @@
 import asyncio
+import os
 import threading
 import time
-import os
-from asyncio import Future, AbstractEventLoop
+from asyncio import AbstractEventLoop, Future
 from typing import Dict
 
 from aio_pika import Message
 from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
-
 from opentelemetry import trace
-from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 from opentelemetry.trace import Tracer, TraceState
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
 from amqp.adapter.backpressure_handler import BackpressureHandler
+from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.data_response_factory import DataResponseFactory
-from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
-from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.serializer import map_as_string
 from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import (
-    DataResponse,
     DataMessage,
+    DataMessageMagicByte,
+    DataResponse,
     MultipartDataMessage,
 )
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
-from amqp.adapter.serializer import map_as_string
 
 
 def collect_trace_info():
@@ -75,12 +75,15 @@ class DataMessageHandler:
         self.file_handler: FileHandler = file_handler
         self.backpressure_handler: BackpressureHandler = backpressure_handler
 
-    async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback_as_thread(self, message: AbstractIncomingMessage):
+        """
+        The callback method that is called when a DataMessage is received from RabbitMQ, starts new Thread.
+        """
         threading.Thread(
-            target=lambda: asyncio.run(self.run_callback_thread(message))
+            target=lambda: asyncio.run(self.inbound_data_message_callback_no_thread(message))
         ).start()
 
-    async def run_callback_thread(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback_no_thread(self, message: AbstractIncomingMessage):
         """
         Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
         Ensures Jaeger trace-id propagation.
@@ -92,19 +95,17 @@ class DataMessageHandler:
         amq_message = await self.reconstitute_data_message(message)
         if amq_message is not None:
             self.ensure_trace_info(amq_message)
-            if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
+            if amq_message.magic() == DataMessageMagicByte.HTTP_REQUEST.value:
                 await self.run_http_request_magic(message, amq_message, self.loop)
             else:
                 logging_error(
                     "Unknown MAGIC value: [%s], don't know how to interpret the message",
-                    amq_message.magic,
+                    amq_message.magic(),
                 )
                 asyncio.run_coroutine_threadsafe(message.nack(requeue=False), self.loop)
                 return
         else:
-            logging_error(
-                "######[%s] No valid AMQ Message present.", message.delivery_tag
-            )
+            logging_error("######[%s] No valid AMQ Message present.", message.delivery_tag)
         self.record_duration(start, message.delivery_tag)
         asyncio.run_coroutine_threadsafe(message.ack(multiple=False), self.loop)
 
@@ -168,9 +169,7 @@ class DataMessageHandler:
             amq_message = await DataMessageFactory.from_stream(
                 message.body, rabbitmq_url=self.rabbit_mq_url, loop=self.loop
             )
-            logging_info(
-                f"######[{delivery_tag}] AMQ Message from stream: {amq_message}"
-            )
+            logging_info(f"######[{delivery_tag}] AMQ Message from stream: {amq_message}")
             if amq_message is not None:
                 extra_data = await self.file_handler.on_message(
                     message=message,
@@ -179,26 +178,24 @@ class DataMessageHandler:
                     loop=self.loop,
                     output_dir=self.output_dir,
                 )
-                logging_info(
-                    f"######[{delivery_tag}] Multipart download result: {extra_data}"
-                )
+                logging_info(f"######[{delivery_tag}] Multipart download result: {extra_data}")
                 amq_message = MultipartDataMessage(amq_message, extra_data)
         if amq_message:
             logging_info(
                 "######[%s] AMQ Message received, ctag=%s, msgId=%s, %s:%s, traceInfo=%s",
                 message.delivery_tag,
                 message.consumer_tag,
-                amq_message.id,
-                amq_message.method,
-                amq_message.path,
-                map_as_string(amq_message.trace_info),
+                amq_message.id(),
+                amq_message.method(),
+                amq_message.path(),
+                map_as_string(amq_message.trace_info()),
             )
         return amq_message
 
     def ensure_trace_info(self, amq_message: DataMessage):
         propagator = TraceContextTextMapPropagator()
         context = propagator.extract(
-            carrier=amq_message.trace_info, context=trace.context_api.get_current()
+            carrier=amq_message.trace_info(), context=trace.context_api.get_current()
         )
         self.tracer.start_span(name="AMQ-Adapter-Consumer", context=context)
         # logging_info("CTX=%s", context)
@@ -207,7 +204,7 @@ class DataMessageHandler:
         # logging.info(" [*] CTX2=%s", context_with_span)
         # propagator.inject(context_with_span, amq_message.trace_info, self.set_text)
         # logging.info(" [***] NEW TI=%s CTX=%s SCOPE=%s",
-        #             map_as_string(amq_message.trace_info), context_with_span, context)
+        #             map_as_string(amq_message.trace_info()), context_with_span, context)
 
     def record_duration(self, start, delivery_tag):
         global last_data_message_time
@@ -229,17 +226,15 @@ class DataMessageHandler:
         if reply_to:
             pika_message: Message = Message(
                 body=DataResponseFactory.serialize(response),
-                correlation_id=response.id,
+                correlation_id=response.id(),
                 content_type=(
-                    response.content_type[0]
-                    if isinstance(response.content_type, list)
-                    else response.content_type
+                    response.content_type()[0]
+                    if isinstance(response.content_type(), list)
+                    else response.content_type()
                 ),
             )
             _res = asyncio.run_coroutine_threadsafe(
-                self.reply_to_exchange.publish(
-                    message=pika_message, routing_key=reply_to
-                ),
+                self.reply_to_exchange.publish(message=pika_message, routing_key=reply_to),
                 self.loop,
             )
             logging_debug(
@@ -257,15 +252,12 @@ class DataMessageHandler:
         :return:
         """
         reply_id = message.correlation_id
-        content_type = message.content_type
         delivery_tag = message.delivery_tag
         debug = f"######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}"
         logging_debug(debug)
         # TODO - figure out implementation of this part, as this is sending reply back either to a service,
         #        or to the user
-        logging_info(
-            f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}"
-        )
+        logging_info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}")
 
         future: Future = self.outstanding.pop(reply_id, None)
         # if reply is None:

amqp/service/amq_service.py

@@ -1,26 +1,24 @@
+import asyncio
 import logging
 import logging.config
 import os
-import asyncio
 from asyncio import Future
 from threading import Thread
 from typing import Optional
 
 from aio_pika.abc import AbstractRobustExchange
 
+from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.backpressure_handler import BackpressureHandler
 from amqp.adapter.data_message_factory import DataMessageFactory
-from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.logging_utils import logging_info, logging_warning
 from amqp.adapter.service_message_factory import ServiceMessageFactory
-
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import DataMessage
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
 from amqp.service.amq_message_handler import DataMessageHandler
 from amqp.service.tracing import initialize_trace
 
-
 logging.basicConfig(level=logging.DEBUG)
 logging.info(f"CWD = {os.getcwd()}")
 logging_conf_path = os.path.join(os.getcwd(), "amqp", "service", "logging.conf")
@@ -66,9 +64,7 @@ class AMQService:
         self.loop.run_forever()
 
     async def init(self, loop, service_adapter):
-        _reply_to_exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(
-            loop
-        )
+        _reply_to_exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(loop)
         backpressure_handler = BackpressureHandler(
             channel=self.rabbit_mq_client.channel,
             loop=loop,
@@ -107,7 +103,7 @@ class AMQService:
             try:
                 await self.rabbit_mq_client.register_inbound_routes(
                     route_mapping,
-                    self.amq_data_message_handler.inbound_data_message_callback,
+                    self.amq_data_message_handler.inbound_data_message_callback_no_thread,
                 )
                 return await self.rabbit_mq_client.register_data_reply_callback(
                     self.amq_data_message_handler.reply_received_callback
@@ -122,7 +118,7 @@ class AMQService:
         return None
 
     def send_message(self, message: DataMessage, future: Future):
-        self.amq_data_message_handler.outstanding[str(message.id)] = future
+        self.amq_data_message_handler.outstanding[str(message.id())] = future
 
     def remove_outstanding_future(self, message_id: str):
         self.amq_data_message_handler.outstanding.pop(message_id, None)

amqp/service/tracing.py

@@ -1,15 +1,14 @@
-from opentelemetry import metrics, trace
-from opentelemetry.sdk.metrics import MeterProvider
-from opentelemetry.sdk.trace import TracerProvider, Tracer
-from opentelemetry.sdk.trace.export import BatchSpanProcessor
-from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
-
-from prometheus_client import start_http_server
-from opentelemetry.exporter.prometheus import PrometheusMetricReader
-from opentelemetry.sdk.resources import SERVICE_NAME, Resource
-
 import os
 
+from opentelemetry import metrics, trace
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.exporter.prometheus import PrometheusMetricReader
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import Tracer, TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from prometheus_client import start_http_server
+
 
 def initialize_trace(app=None, name=None) -> Tracer:
     # Resource can be required for some backends, e.g. Jaeger or Prometheus
@@ -17,9 +16,7 @@ def initialize_trace(app=None, name=None) -> Tracer:
     # The 'name' value is the name shown in Jaeger.
     resource = Resource(
         attributes={
-            SERVICE_NAME: (
-                name if name else os.environ.get("APPLICATION_NAME", "Python App")
-            )
+            SERVICE_NAME: (name if name else os.environ.get("APPLICATION_NAME", "Python App"))
         }
     )
     #

cleverbrag/clever_brag_adapter.py

@@ -0,0 +1,122 @@
+"""
+This module provides the CleverBRAG AMQP Adapter for handling API requests via AMQP.
+The code IS MEANT to be part of CleverBRAG codebase, and has direct dependencies on the CleverBRAG codebase.
+It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the CleverBRAG context.
+"""
+
+from threading import Thread
+from typing import Any, List
+from uuid import UUID
+
+# CleverBrag specific imports
+import core
+from fastapi.routing import APIRoute
+from fastapi.security.http import HTTPAuthorizationCredentials
+from pydantic import BaseModel
+from shared.abstractions import R2RSerializable, User
+
+# AMQP specific imports
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
+from amqp.config.amq_configuration import AMQConfiguration
+from amqp.service.amq_service import AMQService
+
+
+class BRAGArgument:
+    """
+    This class is used to represent a BRAG API input argument. It is uses FastAPI annotation to define the arguments
+    (the argument type) that are passed as input arguments to the BRAG endpoint.
+    """
+
+    def __init__(self, arg: Any):
+        self.arg = arg
+        self.module = arg.__module__
+        self.cls = arg.__name__ if hasattr(arg, "__name__") else str(arg)
+        self.is_r2r_serializable = issubclass(arg, R2RSerializable)
+        self.is_builtin = self.module == "builtins"
+        self.is_uuid = self.cls == "UUID"
+        self.is_optional = self.cls == "Optional"
+        self.is_list = self.cls == "list"
+        self.is_dict = self.cls == "dict"
+        self.is_base_model = issubclass(arg, BaseModel)
+        self.is_nested = self.is_list or self.is_optional or self.is_dict
+        if self.is_nested and hasattr(arg, "__args__"):
+            self.nested = BRAGArgument(getattr(arg, "__args__")[0])
+            if self.is_dict:
+                self.nested = (self.nested, BRAGArgument(getattr(arg, "__args__")[1]))
+        else:
+            self.nested = None
+            self.is_nested = False
+
+    def __repr__(self):
+        return f"{self.module}.{self.cls}" + (f"[{self.nested}]" if self.is_nested else "")
+
+
+def convert_to_argument(arg: BRAGArgument, value: str) -> Any:
+    """
+    Converts the argument to a string representation.
+    """
+    try:
+        if arg.cls == "str":
+            return value
+        elif arg.cls == "bool":
+            return str(value).lower() == "true"
+        elif arg.cls == "float":
+            return float(value)
+        elif arg.cls == "int":
+            return int(value)
+        elif arg.is_uuid:
+            return UUID(value)
+        elif arg.is_optional:
+            return convert_to_argument(arg.nested, value) if arg.nested else None
+        elif arg.is_list:
+            return (
+                [convert_to_argument(arg.nested, v.strip()) for v in value.split(",")]
+                if arg.nested
+                else value.split()
+            )
+        elif arg.is_dict:
+            return (
+                {
+                    convert_to_argument(arg.nested[0], k.strip()): convert_to_argument(
+                        arg.nested[1], v.strip()
+                    )
+                    for k, v in (item.split(":") for item in value.split(","))
+                }
+                if arg.nested
+                else dict(item.split(":") for item in value.split(","))
+            )
+        elif arg.is_base_model:
+            return arg.arg.parse_raw(value)
+        else:
+            print(f"Unsupported argument type: {type(arg)}")
+    except Exception as e:
+        print(f"Error converting argument {arg}: {e}")
+    return None
+
+
+class CleverBragAmqpAdapter(CleverThisServiceAdapter):
+    """
+    CleverBRAG AMQP Adapter for CleverBRAG API. Implements the interface to CleverBRAG API.
+    """
+
+    def __init__(
+        self,
+        amq_configuration: AMQConfiguration,
+        routes: List[APIRoute],
+        auth_provider: core.base.providers.auth.AuthProvider,
+    ):
+        super().__init__(amq_configuration, routes, None)
+        self.auth_provider = auth_provider
+        _amq_service: AMQService = AMQService(amq_configuration, self)
+        self.thread = Thread(target=_amq_service.run)
+        self.thread.start()
+
+    async def get_current_user(self, token: str) -> User:
+        """
+        Overrides the default get_current_user method to use the token from the AMQP message.
+        :param token: JWT token from the AMQP message
+        :return: User object
+        """
+        return await self.auth_provider.auth_wrapper(
+            HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
+        )

cleverbrag/main.py

@@ -0,0 +1,41 @@
+from clever_brag_adapter import BRAGArgument, CleverBragAmqpAdapter
+
+from amqp.config.amq_configuration import AMQConfiguration
+
+
+class R2RApp:
+    pass
+
+
+def create_r2r_app():
+    return R2RApp()
+
+
+if __name__ == "__main__":
+
+    #
+    # r2rapp: R2RApp = await create_r2r_app()
+    #
+    # In the standalone test context, need to start the loop as well. In the actual BRAG context,
+    # use 'await' instead.
+    r2rapp: R2RApp = create_r2r_app()
+
+    adapter: CleverBragAmqpAdapter = CleverBragAmqpAdapter(
+        AMQConfiguration("./application.properties.local"), r2rapp.app.routes, r2rapp.providers.auth
+    )
+
+    for method, routes in adapter.routes.items():
+        print(f"Method: {method}")
+        for _route in routes:
+            print(
+                f"  Route: {_route.path} -> {_route.endpoint.__module__}.{_route.endpoint.__name__}"
+            )
+            _args = getattr(_route.endpoint, "__annotations__", {})
+            _verified_args = []
+            for arg in _args:
+                _arg = BRAGArgument(_args[arg])
+                _verified_args.append(_arg)
+            print(f"    Arguments: {', '.join(map(str, _verified_args))}")
+            print("---------------------------")
+    print("Press ^C to exit...")
+    adapter.thread.join()

cleverswarm/clever_swarm_adapter.py

@@ -5,129 +5,51 @@ It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the C
 """
 
 import importlib
-import pathlib
-import re
-from collections import defaultdict
 from threading import Thread
-from typing import Optional, List, Dict
-from aiohttp import ClientSession
-from fastapi.routing import APIRoute
-from fastapi.security import OAuth2PasswordRequestForm
-from starlette.datastructures import UploadFile, Headers
+from typing import Any, List, Optional
 
 # CleverSwarm specific imports
 import core.deps
-from rest.v0 import api
+from fastapi.routing import APIRoute
 from schemas.user_schema import UserSchema
 
-from amqp.adapter.data_parser import AMQDataParser
-
-# AMQP specific imports
-from amqp.adapter.data_response_factory import AMQResponseFactory
-from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQResponse
 from amqp.service.amq_service import AMQService
 
-
-# Suffix for the method name with serialization wrapper, prefer use this one over the orig endpoint
 PREFERRED_SUFFIX = "_json"
 
 
-def clone_and_adapt_route(original_route: APIRoute) -> APIRoute:
+# =================================================================================================
+# =========================== S u p p o r t i n g   f u n c t i o n s =============================
+# =================================================================================================
+def get_callable_for_name(module_name: str, function_name: str) -> Optional[callable]:
     """
-    Creates a clone of an APIRoute. It checks if there's preferred endpoint with json serialization.
-    It also adjusts the path regex to ensure it matches the path in different API versions.
+    Get the function/Callable for the module_name.function_name if it exists in the module, else return None.
     """
-
-    # Check if the route has a preferred endpoint with json serialization
-    _preferred_endpoint = get_prefered_endpoint(
-        getattr(original_route.endpoint, "__module__"),
-        original_route.endpoint.__name__ + PREFERRED_SUFFIX,
-    )
-    # Need to adapt also the path regex to ensure it matches the path in different API versions
-    # Ensures the leading '^' is followed by any match and there's no '/' before final '$'
-    _new_regex_string = (
-        "^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$"
-    )
-    _new_route: APIRoute = APIRoute(
-        path=original_route.path,
-        endpoint=(
-            _preferred_endpoint if _preferred_endpoint else original_route.endpoint
-        ),
-        response_model=original_route.response_model,
-        status_code=original_route.status_code,
-        tags=original_route.tags.copy() if original_route.tags else None,
-        dependencies=(
-            original_route.dependencies.copy() if original_route.dependencies else None
-        ),
-        summary=original_route.summary,
-        description=original_route.description,
-        response_description=original_route.response_description,
-        responses=original_route.responses.copy() if original_route.responses else None,
-        deprecated=original_route.deprecated,
-        name=original_route.name,
-        methods=original_route.methods.copy() if original_route.methods else None,
-        operation_id=original_route.operation_id,
-        response_model_include=original_route.response_model_include,
-        response_model_exclude=original_route.response_model_exclude,
-        response_model_by_alias=original_route.response_model_by_alias,
-        response_model_exclude_unset=original_route.response_model_exclude_unset,
-        response_model_exclude_defaults=original_route.response_model_exclude_defaults,
-        response_model_exclude_none=original_route.response_model_exclude_none,
-        include_in_schema=original_route.include_in_schema,
-        response_class=original_route.response_class,
-        dependency_overrides_provider=original_route.dependency_overrides_provider,
-        callbacks=original_route.callbacks.copy() if original_route.callbacks else None,
-        openapi_extra=(
-            original_route.openapi_extra.copy()
-            if original_route.openapi_extra
-            else None
-        ),
-        generate_unique_id_function=original_route.generate_unique_id_function,
-    )
-    _new_route.path_regex = re.compile(_new_regex_string)
-    return _new_route
-
-
-def get_prefered_endpoint(module_name: str, function_name: str) -> Optional[callable]:
-    """Get the function if it exists in the module, else return None."""
     try:
         _module = importlib.import_module(module_name)
-        _prefered_endpoint = getattr(_module, function_name)
-        return _prefered_endpoint if callable(_prefered_endpoint) else None
+        _preferred_endpoint = getattr(_module, function_name)
+        return _preferred_endpoint if callable(_preferred_endpoint) else None
     except Exception:
         return None
 
 
-def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]:
-    """
-    Groups API routes by their HTTP method. This is important to ensure that the URLPath is matched
-    correctly, as different methods may share the same URLPath (e.g. GET, POST PUT, DELETE).
-    """
-    _method_to_routes = defaultdict(
-        list
-    )  # Automatically initializes new lists for new keys
-    for _route in api_routes:
-        if (
-            _route.methods
-        ):  # Ensure methods exist (should always be true for valid routes)
-            for method in _route.methods:
-                _method_to_routes[method].append(clone_and_adapt_route(_route))
-
-    return dict(_method_to_routes)
-
-
+# =================================================================================================
+# ======================== C l e v e r S w a r m   A M Q    A d a p t e r  ========================
+# =================================================================================================
 class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
     """
-    CleverSwarm AMQP Adapter for CleverSwarm API. Implements the interface to CleverSwarm API.
+    CleverSwarm AMQP Adapter for CleverSwarm API. Implements the CleverSwarm specific logic that overrides
+    common logic defined by AMQ Adapter library. The specific logic here is:
+
+    1. Extract subject from JWT token and convert it to UserSchema to avail authorized user argument
+    2. Find JSON wrapper for endpoint Callable, and if exists, use it instead of the original endpoint.
+
     """
 
-    def __init__(
-        self, amq_configuration: AMQConfiguration, session: ClientSession = None
-    ):
-        super().__init__(amq_configuration, session)
-        self.routes = group_routes_by_method(api_routes=api.api_router.routes)
+    def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]):
+        super().__init__(amq_configuration, routes=routes, session=None)
         _amq_service: AMQService = AMQService(amq_configuration, self)
         self.thread = Thread(target=_amq_service.run)
         self.thread.start()
@@ -135,168 +57,25 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
     async def get_current_user(self, token: str) -> UserSchema:
         """
         Overrides the default get_current_user method to use the token from the AMQP message.
+        In CleverSwarm context, it means simply to call provided get_current_user function.
+
         :param token: JWT token from the AMQP message
+
         :return: UserSchema object
         """
         return await core.deps.get_current_user(token)
 
-    # ==================================================================================================
-    # =====================  C l e v e r S w a r m   A P I   m e t h o d s  ============================
-    # ==================================================================================================
+    def get_endpoint(self, endpoint: Any) -> Any:
+        """
+        If the specific endpoint has a matching JSON wrapper (identified as {endpoint.name}_json ),
+        then use it as preferred way to invoke the endpoint.
 
-    async def get(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-        """
-        Handles the GET requests for the CleverSwarm API.
-        :param message: message from the AMQP that contains the GET request
-        :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
-        """
-        return await self.handle_no_body_payload(message, auth_user, "GET")
+        :param endpoint: the Callable for the endpoint
 
-    async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
+        :return: new Callable representing the JSON wrapper endpoint, if it exists or the original Callable.
         """
-        Handles the PUT requests for the CleverSwarm API.
-        :param message: message from the AMQP that contains the PUT request
-        :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
-        """
-        return await self.handle_body_payload(message, auth_user, "PUT")
-
-    async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-        """
-        Handles the POST requests for the CleverSwarm API.
-        :param message: message from the AMQP that contains the POST request, including body and headers
-        :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
-        """
-        return await self.handle_body_payload(message, auth_user, "POST")
-
-    async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-        """
-        Handles the HEAD requests for the CleverSwarm API.
-        """
-        return await self.handle_no_body_payload(message, auth_user, "HEAD")
-
-    async def delete(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-        """
-        Handles the DELETE requests for the CleverSwarm API.
-        """
-        return await self.handle_no_body_payload(message, auth_user, "DELETE")
-
-    async def options(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-        """
-        Handles the OPTIONS requests for the CleverSwarm API.
-        """
-        return await self.handle_no_body_payload(message, auth_user, "OPTIONS")
-
-    async def patch(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-        """
-        Handles the PATCH requests for the CleverSwarm API.
-        """
-        return await self.handle_no_body_payload(message, auth_user, "PATCH")
-
-    async def handle_no_body_payload(
-        self, message: AMQMessage, auth_user: UserSchema, method: str
-    ) -> AMQResponse:
-        """
-        Handles the requests that do not have a body payload.
-        :param message: message from the AMQP that contains the request
-        :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
-        """
-        _routes = self.routes.get(method, [])
-        _path, _args = AMQDataParser.parse_get_url(message.path)
-        for _route in _routes:
-            _match = _route.path_regex.match(_path)
-            if _match:
-                _args.update(_match.groupdict())
-                _args["authenticated_user"] = auth_user
-                # If the path matches, call the corresponding function
-                return await CleverThisServiceAdapter.call_cleverthis_api(
-                    message,
-                    _args,
-                    api_method=_route.endpoint,
-                    success_code=_route.status_code,
-                    loop=self.loop,
-                )
-
-        return AMQResponseFactory.of(
-            message.id,
-            404,
-            "text/plain",
-            str(f"Destination location [{message.path}] does not exists").encode(
-                "utf-8"
-            ),
-            message.trace_info,
-        )
-
-    async def handle_body_payload(
-        self, message: AMQMessage, auth_user: UserSchema, method: str
-    ) -> AMQResponse:
-        """
-        Handles the requests that have a body payload, including Multipart data.
-        :param message: message from the AMQP that contains the request
-        :param auth_user: user authenticated by the API Gateway, as UserSchem
-        :param method: HTTP method (POST, PUT, etc.)
-        :return: AMQResponse to be sent back to the API Gateway via AMQP
-        """
-        _form_data = AMQDataParser.parse_request_body(message)
-        _routes = self.routes.get(method, [])
-        _path, _query_args = AMQDataParser.parse_get_url(message.path)
-        for _route in _routes:
-            _match = _route.path_regex.match(_path)
-            if _match:
-                # If the path matches, call the corresponding function
-                _form_data.update(_match.groupdict())
-                _form_data.update(_query_args)
-                for _body_param in _route.dependant.body_params:
-                    if _body_param.name in _form_data:
-                        _file_data = _form_data[_body_param.name]
-                        actual_filepath = message.extra_data[_body_param.name]
-                        _form_data[_body_param.name] = (
-                            _body_param.type_.__name__ == "UploadFile"
-                            and UploadFile(
-                                file=pathlib.Path(actual_filepath).open("rb"),
-                                size=_file_data["size"],
-                                filename=_file_data["filename"],
-                                headers=Headers(
-                                    {"Content-Type": _file_data["mediaType"]}
-                                ),
-                            )
-                            or _form_data[_body_param.name]
-                        )
-
-                _verified_args = {}
-                _required_args = getattr(_route.endpoint, "__annotations__", {})
-                for arg in _required_args:
-                    if arg in _form_data:
-                        _verified_args[arg] = _form_data[arg]
-                    else:
-                        # needed for /login  input
-                        if arg == "authenticated_user":
-                            _verified_args["authenticated_user"] = auth_user
-                        if _required_args[arg].__name__ == "OAuth2PasswordRequestForm":
-                            try:
-                                _verified_args[arg] = OAuth2PasswordRequestForm(
-                                    **_form_data
-                                )
-                            except Exception as e:
-                                print(f"Error: {e}")
-
-                return await CleverThisServiceAdapter.call_cleverthis_api(
-                    message,
-                    _verified_args,
-                    api_method=_route.endpoint,
-                    success_code=_route.status_code if _route.status_code else 200,
-                    loop=self.loop,
-                )
-
-        return AMQResponseFactory.of(
-            message.id,
-            404,
-            "text/plain",
-            str(f"Destination location [{message.path}] does not exists").encode(
-                "utf-8"
-            ),
-            message.trace_info,
+        _preferred_endpoint = get_callable_for_name(
+            getattr(endpoint, "__module__"),
+            endpoint.__name__ + PREFERRED_SUFFIX,
         )
+        return _preferred_endpoint if _preferred_endpoint is not None else endpoint

cleverswarm/main.py

@@ -1,7 +1,7 @@
-from amqp.config.amq_configuration import AMQConfiguration
-
 from clever_swarm_adapter import CleverSwarmAmqpAdapter
 
+from amqp.config.amq_configuration import AMQConfiguration
+
 if __name__ == "__main__":
     adapter = CleverSwarmAmqpAdapter(AMQConfiguration("./application.properties.local"))
     print("Press ^C to exit...")

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.23-dev"
-description = "The Python implementation of the AMQ Adaptor for CleverMicro"
+version = "0.2.31"
+description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -19,15 +19,16 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika~=9.5",
-    "aiohttp~=3.11",
-    "aio_pika~=9.5",
-    "fastapi~=0.114",
+    "aio-pika",
+    "aiohttp",
+    "fastapi",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
+    "pika",
+    "psutil",
     "python_multipart"
 ]
 
@@ -35,11 +36,18 @@ dependencies = [
 where = ["."]
 include = ["amqp*"]
 
+[tool.black]
+line-length = 100
+
 [project.optional-dependencies]
 dev = [
     "pytest>=7.0",
     "pytest-cov>=4.0",
+    "pytest-asyncio",
     "build",
-    "twine"
+    "twine",
+    "black",
+    "isort",
+    "flake8",
+    "pre-commit"
 ]
-

tests/adapter/test_cleverthis_service_adapter.py

@@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 from aiohttp import ClientResponse
 
-from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
+from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import DataMessage, DataResponse
 
@@ -18,7 +18,7 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         # Create adapter instance
         self.adapter = CleverThisServiceAdapter(
-            amq_configuration=self.mock_amq_config, session=self.mock_session
+            amq_configuration=self.mock_amq_config, routes=[], session=self.mock_session
         )
 
     def test_init(self):
@@ -56,7 +56,7 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         # Test when authentication is required but not provided
         response = await self.adapter.on_message(mock_message)
-        self.assertEqual(response.response_code, 401)
+        self.assertEqual(response.response_code(), "401")
         mock_get_user.assert_not_called()
 
     @patch.object(CleverThisServiceAdapter, "get_current_user")
@@ -91,9 +91,7 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
         mock_message.trace_info = {}
 
         mock_response = MagicMock(spec=DataResponse)
-        with patch.object(
-            self.adapter, "post", return_value=mock_response
-        ) as mock_post:
+        with patch.object(self.adapter, "post", return_value=mock_response) as mock_post:
             response = await self.adapter.on_message(mock_message)
             self.assertEqual(response, mock_response)
             mock_get_user.assert_not_called()
@@ -139,7 +137,7 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         response = await self.adapter.get(mock_message, None)
 
-        self.assertEqual(response.response_code, 200)
+        self.assertEqual(response.response_code(), "200")
         self.mock_session.get.assert_called_once_with(
             "http://testhost:8080/api/test",
             headers={"X-Test": "value", "trace-id": "123"},
@@ -155,7 +153,7 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         response = await self.adapter.get(mock_message, None)
 
-        self.assertEqual(response.response_code, 404)
+        self.assertEqual(response.response_code(), "404")
 
     async def test_put_with_form_handling(self):
         mock_message = MagicMock(spec=DataMessage)
@@ -174,12 +172,12 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "handle_possible_form",
-            wraps=self.adapter.handle_possible_form,
+            "_authorize_forward_request",
+            wraps=self.adapter._authorize_forward_request,
         ) as mock_handler:
             response = await self.adapter.put(mock_message, None)
 
-            self.assertEqual(response.response_code, 200)
+            self.assertEqual(response.response_code(), "200")
             mock_handler.assert_called_once()
 
     async def test_post_with_form_handling(self):
@@ -199,12 +197,12 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "handle_possible_form",
-            wraps=self.adapter.handle_possible_form,
+            "_authorize_forward_request",
+            wraps=self.adapter._authorize_forward_request,
         ) as mock_handler:
             response = await self.adapter.post(mock_message, None)
 
-            self.assertEqual(response.response_code, 201)
+            self.assertEqual(response.response_code(), "201")
             mock_handler.assert_called_once()
 
     async def test_head_not_implemented(self):

tests/adapter/test_data_message_factory.py

@@ -10,25 +10,21 @@ class DataMessageFactoryTest(TestCase):
         msg = DataMessageFactory.from_bytes(raw.encode("utf-8"))
         self.assertIsNotNone(msg.id)
         self.assertEqual(
-            msg.id.as_string(),
+            msg.id().as_string(),
             "64F3B5AF4D00000A4000".lower(),
             "SnowflakeID incorrectly parsed.",
         )
         self.assertEqual(
-            msg.headers["Content-Type"][0],
+            msg.headers()["Content-Type"][0],
             "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v",
         )
 
     def test_generate_snowflake_id(self):
         _id = DataMessageFactory.get_instance(1).generate_snowflake_id()
-        self.assertEqual(
-            "00001000", _id.as_string()[-8:], "Instance ID not set correctly"
-        )
+        self.assertEqual("00001000", _id.as_string()[-8:], "Instance ID not set correctly")
 
         _id = DataMessageFactory(2).generate_snowflake_id()
-        self.assertEqual(
-            "00002000", _id.as_string()[-8:], "Instance ID not set correctly"
-        )
+        self.assertEqual("00002000", _id.as_string()[-8:], "Instance ID not set correctly")
 
         _f = DataMessageFactory(127)
         _i0 = _f.generate_snowflake_id()
@@ -65,9 +61,9 @@ class DataMessageFactoryTest(TestCase):
             {"k1": ["v1"], "k2": ["v2"]},
             "Quick brown fox",
         )
-        self.assertEqual("localhost", _req.domain, "Domain not set correctly")
-        self.assertEqual("/test/me", _req.path, "Context path not set correctly")
-        self.assertEqual("v1", _req.headers["k1"][0], "Headers not set correctly")
+        self.assertEqual("localhost", _req.domain(), "Domain not set correctly")
+        self.assertEqual("/test/me", _req.path(), "Context path not set correctly")
+        self.assertEqual("v1", _req.headers()["k1"][0], "Headers not set correctly")
         self.assertEqual(b"Quick brown fox", _req.body(), "Body not correctly decoded")
 
     def test_serialize(self):
@@ -82,11 +78,11 @@ class DataMessageFactoryTest(TestCase):
         _ser = _f.serialize(_req)
         self.assertEqual(65, _ser[2])
         _deser = _f.from_bytes(_ser)
-        self.assertEqual(_req.id.as_string(), _deser.id.as_string())
-        self.assertEqual("localhost", _deser.domain)
-        self.assertEqual("/test/me", _deser.path)
+        self.assertEqual(_req.id().as_string(), _deser.id().as_string())
+        self.assertEqual("localhost", _deser.domain())
+        self.assertEqual("/test/me", _deser.path())
         self.assertEqual(b"Quick brown fox", _deser.body())
-        self.assertEqual("v2", _deser.headers["k2"][0])
+        self.assertEqual("v2", _deser.headers()["k2"][0])
 
     def test_amq_message_routing_key(self):
         _f = DataMessageFactory(35)
@@ -124,6 +120,4 @@ class DataMessageFactoryTest(TestCase):
             "Quick brown fox",
         )
         _as_str = _f.to_string(_req)
-        self.assertGreater(
-            _as_str.index("domain='localhost'"), 0, "Expected content not found"
-        )
+        self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")

tests/adapter/test_data_response_factory.py

@@ -2,6 +2,7 @@ from unittest import TestCase
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.data_response_factory import DataResponseFactory
+from amqp.model.model import DataMessageMagicByte
 from amqp.model.snowflake_id import SnowflakeId
 
 
@@ -11,8 +12,8 @@ class TestDataResponseFactory(TestCase):
         _f = DataResponseFactory()
         _g = DataMessageFactory(111)
         _resp = _f.create_async_response_message(_g.generate_snowflake_id())
-        self.assertEqual(200, _resp.response_code)
-        self.assertEqual(b"OK", _resp.response)
+        self.assertEqual("200", _resp.response_code())
+        self.assertEqual(b"OK", _resp.body())
 
     def test_serialize(self):
         _f = DataResponseFactory()
@@ -22,7 +23,7 @@ class TestDataResponseFactory(TestCase):
         _ser = _f.serialize(_resp)
         self.assertEqual(
             2,
-            str(_ser[2:]).index(_f.MAGIC_BYTE_HTTP_REQUEST + _snowflakeId.as_string()),
+            str(_ser[2:]).index(DataMessageMagicByte.HTTP_REQUEST.value + _snowflakeId.as_string()),
             "Magic byte and SnowflakeId not found",
         )
         self.assertGreater(str(_ser[2:]).index("|e=|"), 2)
@@ -35,6 +36,6 @@ class TestDataResponseFactory(TestCase):
         _resp = _f.create_async_response_message(_snowflakeId)
         _ser = _f.serialize(_resp)
         _deser = _f.from_bytes(_ser)
-        self.assertEqual(_snowflakeId.as_string(), _deser.id)
-        self.assertEqual(200, _deser.response_code)
-        self.assertEqual(b"OK", _deser.response)
+        self.assertEqual(_snowflakeId.as_string(), _deser.id())
+        self.assertEqual("200", _deser.response_code())
+        self.assertEqual(b"OK", _deser.body())

tests/router/test_route_database.py

@@ -2,8 +2,8 @@ from unittest import TestCase
 
 import pytest
 
-from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.amq_route_factory import AMQRouteFactory
+from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.model.model import AMQRoute
 from amqp.router.route_database import RouteDatabase
 
@@ -80,7 +80,7 @@ class TestRouteDatabase(TestCase):
         message = DataMessageFactory.get_instance(111).create_request_message(
             "POST", "clever3-book.localhost", "/aa/bb?cc=d", {}, ""
         )
-        route = self._database.find_route(message.domain, message.path)
+        route = self._database.find_route(message.domain(), message.path())
         self.assertIsNotNone(route)
         self.assertEqual(ROUTE_3, route.as_string())
         route = self._database.find_route("clever3.book.localhost", "/three-hot-dogs")

tools/create_exchange.py

@@ -0,0 +1,90 @@
+import asyncio
+import aio_pika
+import os
+import argparse
+import sys
+
+from aio_pika.abc import AbstractRobustExchange
+
+
+async def create_exchange(loop, rabbit_user, rabbit_password, rabbit_host):
+    """
+    Asynchronously connects to RabbitMQ, declares a direct exchange,
+    and prints the result.  Handles potential connection errors.
+    """
+    exchange_name = "cleverthis.clevermicro.management"
+    connection = None  # Keep track of the connection
+
+    try:
+        # Construct the connection URI from environment variables
+        connection_uri = f"amqp://{rabbit_user}:{rabbit_password}@{rabbit_host}/"
+
+        # Attempt to establish the connection
+        connection = await aio_pika.connect_robust(connection_uri, loop=loop)
+        print(f"Connected to RabbitMQ: {connection_uri}")
+
+        # Create a channel
+        channel = await connection.channel()
+
+        # Declare the exchange
+        exchange: AbstractRobustExchange = await channel.declare_exchange(
+            exchange_name, aio_pika.ExchangeType.DIRECT
+        )
+        print(f"Exchange '{exchange.name}' created successfully.")
+
+    except aio_pika.exceptions.AMQPConnectionError as e:
+        print(f"Error: Could not connect to RabbitMQ.  Details: {e}")
+        sys.exit(1)  # Exit with an error code
+    except aio_pika.exceptions.AMQPChannelError as e:
+        print(f"Error creating channel or exchange: {e}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+        sys.exit(1)
+    finally:
+        # Ensure the connection is closed, even if errors occur
+        if connection:
+            await connection.close()
+            print("Connection to RabbitMQ closed.")
+
+def main():
+    """
+    Main function to parse command-line arguments, retrieve RabbitMQ
+    credentials from environment variables, and run the
+    async exchange creation.
+    """
+    # Set up argument parsing
+    parser = argparse.ArgumentParser(description="Create a RabbitMQ exchange.")
+    parser.add_argument(
+        "--host",
+        default="127.0.0.1",
+        help="RabbitMQ host (default: localhost)",
+    )
+
+    # Parse the arguments
+    args = parser.parse_args()
+    rabbit_host = args.host
+
+    # Get RabbitMQ credentials from environment variables
+    rabbit_user = os.environ.get("RABBIT_MQ_USER")
+    rabbit_password = os.environ.get("RABBIT_MQ_PASSWORD")
+
+    if not rabbit_user or not rabbit_password:
+        print(
+            "Error: RabbitMQ credentials not found in environment variables.\n"
+            "Please set RABBIT_MQ_USER and RABBIT_MQ_PASSWORD."
+        )
+        sys.exit(1)  # Exit with an error code
+
+    # Create the event loop
+    loop = asyncio.get_event_loop()
+
+    # Run the asynchronous exchange creation
+    loop.run_until_complete(create_exchange(loop, rabbit_user, rabbit_password, rabbit_host))
+
+    # Close the loop
+    loop.close()
+
+
+if __name__ == "__main__":
+    main()