fix(General): Fix deps version so cleverswarm can install it

.flake8

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

.pre-commit-config.yaml

@@ -1,21 +0,0 @@
-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,7 +4,6 @@ import time
 from asyncio import AbstractEventLoop
 from threading import Thread
 
-import psutil
 from aio_pika import Message
 from aio_pika.abc import AbstractRobustChannel
 
@@ -47,40 +46,6 @@ 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
@@ -104,7 +69,6 @@ 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"""
@@ -155,90 +119,67 @@ class BackpressureHandler:
                 self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
 
     def backpressure_monitor_loop(self):
-        _time_window_millis = self.config.backpressure.time_window
+        _time_window_sec = self.config.backpressure.time_window
-        _idle_duration_millis = self.config.backpressure.idle_duration
+        _idle_duration_millis = 1000 * 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:
-                _current_cpu_usage = psutil.cpu_percent(interval=None)
+                logging_info(
-                self.average_cpu_usage.add(_current_cpu_usage)
+                    "Backpressure: Monitor loop, current=%s",
-                _average_cpu_usage = self.average_cpu_usage.get_average()
+                    self.current_parallel_executions,
-                _now = time.time()
+                )
-                # logging_info(
+                logging_info(
-                #    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
+                    "Backpressure: Last data message time=%s, eventTime=%s",
-                #    _current_cpu_usage,
+                    self.last_data_message_time,
-                #    _average_cpu_usage,
+                    self.last_backpressure_event_time,
-                #    _cpu_usage_state,
+                )
-                #    self.last_data_message_time,
+                logging_info(
-                #    self.last_backpressure_event,
+                    "Backpressure: Last backpressure event=%s",
-                #    self.last_backpressure_event_time,
+                    self.last_backpressure_event,
-                # )
+                )
-                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 (
-                    _cpu_usage_state == CPU_OVERLOAD
+                    self.current_parallel_executions >= self.parallel_workers
-                    and _now - _cpu_usage_changed > _overload_duration_millis
+                    and time.time() - self.last_backpressure_event_time
-                    and (
+                    > _time_window_sec
-                        _now - self.last_backpressure_event_time > _time_window_millis
+                    and self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
-                        or self.last_backpressure_event != ScalingRequestAlert.OVERLOAD
-                    )
                 ):
                     # Trigger the overload event
                     await self.handle_backpressure_overload_event()
-                    self.last_backpressure_event_time = _now
+                    self.last_backpressure_event_time = time.time()
                     self.last_backpressure_event = ScalingRequestAlert.OVERLOAD
                 elif (
-                    _cpu_usage_state == CPU_IDLE
+                    self.current_parallel_executions == 0
-                    and _now - _cpu_usage_changed > _idle_duration_millis
+                    and time.time() - self.last_data_message_time
-                    and _now - self.last_data_message_time > _idle_duration_millis
+                    >= _idle_duration_millis
-                    and (
+                    and self.last_backpressure_event != ScalingRequestAlert.IDLE
-                        _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 = _now
+                    self.last_backpressure_event_time = time.time()
                 else:
                     # Trigger update event
-                    if _now - self.last_backpressure_event_time > _time_window_millis:
+                    if (
+                        self.current_parallel_executions > 0
+                        and time.time() - self.last_backpressure_event_time
+                        > _time_window_sec
+                    ):
                         # Trigger the update event
                         _scaling_request: ScaleRequestV1 = ScaleRequestV1(
                             self.swarm_service_id,
                             self.swarm_task_id,
-                            100,
+                            self.parallel_workers,
-                            _average_cpu_usage,
+                            self.parallel_workers - self.current_parallel_executions,
                             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 = _now
+                        self.last_backpressure_event_time = time.time()
                         self.last_backpressure_event = ScalingRequestAlert.UPDATE
-                await asyncio.sleep(_monitor_interval)
+                await asyncio.sleep(_time_window_sec)
 
         _loop.run_until_complete(_backpressure_monitor())
 
@@ -280,11 +221,15 @@ 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:
@@ -312,4 +257,6 @@ 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,36 +6,53 @@ 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 collections import defaultdict
+from typing import Dict, List, Callable, Coroutine, Any
-from typing import Any, Callable, Coroutine, Dict, List
 
 from aio_pika.abc import (
-    AbstractExchange,
     AbstractRobustChannel,
     AbstractRobustConnection,
+    AbstractExchange,
 )
-from aiohttp import ClientResponse, ClientSession
+from aiohttp import ClientSession, ClientResponse
-from fastapi import HTTPException, UploadFile
+from dataclasses import dataclass
-from fastapi.routing import APIRoute
+from fastapi import HTTPException
-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_debug, logging_error, logging_info
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import AMQErrorMessage, AMQMessage, DataMessage, DataResponse
+from amqp.model.model import 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,
@@ -44,16 +61,15 @@ 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
+    The file content itself is transmitted through the named stream (which is implemented as temporary dedicated queue.
-    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
+        # Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with correct event loop
-        # correct event loop. This is needed because the RabbitMQ API is not thread-safe and needs
+        # This is needed because the RabbitMQ API is not thread-safe and needs to be called in the context of
-        # to be called in the context of event loop the first RabbitMQ connection was created with.
+        # the 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)
@@ -68,10 +84,9 @@ 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
+    # All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached to parent
-    # to parent event loop, and need to execute at that event loop, not the current one
+    # 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
+    # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
-    # and the coroutine will never execute
     _channel, _exchange, _routing_key = await await_result(
         asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop)
     )
@@ -80,7 +95,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
-    asyncio.run_coroutine_threadsafe(_channel.close(), loop)
+    _future = asyncio.run_coroutine_threadsafe(_channel.close(), loop)
     return json.dumps(
         {
             "files": [
@@ -106,22 +121,19 @@ 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
+    A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the
-    them to the appropriate CleverThisService API endpoint.
+    appropriate CleverThisService API endpoint.
-    IMPORTANT: This class is not intended to be used directly. It should be subclassed for
+    IMPORTANT: This class is not intended to be used directly. It should be subclassed for each CleverThis service
-    each CleverThis service and override the methods to handle the specific API calls.
+    and override the methods to handle the specific API calls.
     """
 
     def __init__(
-        self,
+        self, amq_configuration: AMQConfiguration, session: ClientSession = None
-        amq_configuration: AMQConfiguration,
-        routes: List[APIRoute],
-        session: ClientSession = None,
     ):
         """
-        Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP
+        Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP session.
-        session. The HTTP session is for the case when the adapter uses loose coupling with
+        The HTTP session is for the case when the adapter uses loose coupling with the CleverThis service via
-        the CleverThis service via HTTP REST API.
+        HTTP REST API.
         """
         self.amq_configuration = amq_configuration
         self.session = session
@@ -131,79 +143,6 @@ 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:
         """
@@ -218,11 +157,9 @@ 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
+        To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
-        on the token, in format appropriate for the service.
+        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
@@ -236,9 +173,11 @@ 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]
@@ -250,8 +189,12 @@ 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"):
+        if (
-            method = message.method().lower()
+            _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":
@@ -263,297 +206,114 @@ 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_error_message(
+            _amq_response = DataResponseFactory.of(
-                message.id(),
+                message.id, 401, "text/plain", b"Unauthorized", message.trace_info
-                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
 
-    async def _authorize_forward_request(
+    # ==================================================================================================
-        self, message: DataMessage, request_coroutine, auth_user: Any
+    # =====================  C l e v e r T h i s    A P I    m e t h o d s  ============================
-    ) -> DataResponse:
+    # ==================================================================================================
+
+    async def get(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
-        This method is used to authorize the request and forward it to the CleverThis Service API.
+        Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        Used only for loose / 3rd party mode of coupling with the Service
+        It will attempt to call the REST endpoint directly using HTTP protocol.
-        """
-        # 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
         """
 
-        if self.session is None:
+        # unmatched path - try to invoke the service directly on this path
-            return await self.handle_no_body_payload(message, auth_user, "GET")
+        url = f"http://{self.service_host}:{self.service_port}{message.path}"
-
+        headers = {**message.headers, **message.trace_info}
-        # 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}"
         )
-        _http_resp: ClientResponse = await self.session.get(url, headers=headers)
+        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,
+            )
         return DataResponseFactory.of(
-            message.id(),
+            message.id,
-            _http_resp.status,
+            404,
-            _http_resp.content_type,
+            "text/plain",
-            await _http_resp.read(),
+            str("Destination location does not exists").encode("utf-8"),
-            message.trace_info(),
+            message.trace_info,
         )
 
-    async def put(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+    async def put(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
-        Handles the PUT requests for the CleverThis API. Uses path, path params, query string and
+        Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        body payload to determine the endpoint method to invoke and its input arguments
+        It will attempt to call the REST endpoint directly using HTTP protocol.
-        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
         """
-        if self.session is None:
+        return await self.handle_possible_form(
-            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()}",
+                f"http://{self.service_host}:{self.service_port}{message.path}",
-                data=message.body(),
+                data=message.body,
-                headers={**message.headers(), **message.trace_info()},
+                headers={**message.headers, **message.trace_info},
             ),
             auth_user,
         )
 
-    async def post(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+    async def post(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
-        Handles the POST requests for the CleverThis API. Uses path, path params, query string and
+        Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        body payload to determine the endpoint method to invoke and its input arguments
+        It will attempt to call the REST endpoint directly using HTTP protocol.
-        In loose coupling (self.session is not None), attempt to call the REST endpoint directly
+        :param message: message from the AMQP that contains the POST request, including body and headers
-        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
         """
-        if self.session is None:
+        return await self.handle_possible_form(
-            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()}",
+                f"http://{self.service_host}:{self.service_port}{message.path}",
-                data=message.body(),
+                data=message.body,
-                headers={**message.headers(), **message.trace_info()},
+                headers={**message.headers, **message.trace_info},
             ),
             auth_user,
         )
 
-    async def head(self, message: AMQMessage, auth_user: Any) -> DataResponse:
+    async def handle_possible_form(
+        self, message: DataMessage, request_coroutine, auth_user: Any
+    ) -> 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. In tight coupling (self.session is None),
+        Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
-        use path, path params and query string to determine the endpoint method to invoke.
+        It will attempt to call the REST endpoint directly using HTTP protocol.
-        In loose coupling (self.session is not None), HEAD call is not supported.
+        :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
-        :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:
+    async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse:
         """
-        Handles the DELETE requests for the CleverThis API.
+        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 AMQP that contains the DELETE request, including body & headers
+        :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 UserSchema
+        :param auth_user: user authenticated by the API Gateway, as UserSchem
-
         :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:
-        """
-        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
-        """
-        _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 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 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
-        """
-        _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(
         message: AMQMessage,
@@ -563,15 +323,13 @@ class CleverThisServiceAdapter:
         loop: AbstractEventLoop | None = None,
     ) -> DataResponse:
         """
-        Handles POST and PUT requests, which typically contain possibly complex payload data,
+        Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary.
-        passed as a dictionary. Invokes the provided API method with the given arguments and
+        Invokes the provided API method with the given arguments and authenticated user,
-        authenticated user, and returns the response as DataResponse.
+        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:
@@ -587,19 +345,21 @@ class CleverThisServiceAdapter:
                     elif arg.annotation == float:
                         _verified_args[arg.name] = float(args[arg.name])
                     elif arg.annotation == bool:
-                        _verified_args[arg.name] = serializer.str_to_bool(args[arg.name])
+                        _verified_args[arg.name] = 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(
-                "INVOKE ENDPOINT: {%s}, ARGS: {%s}",
+                f"INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}"
-                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 = (
@@ -611,9 +371,6 @@ 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"
@@ -624,56 +381,44 @@ 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):
-                return DataResponseFactory.of_error_message(
+                _content = '{"error": "' + _content + '"}'
-                    message.id(),
+            logging_error(f"Service REST endpoint invocation failed: {http_error}")
-                    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):
-                return DataResponseFactory.of_error_message(
+                _content = '{"error": "' + _content + '"}'
-                    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 datetime import datetime, timezone
 from io import BytesIO
+from datetime import datetime, timezone
 from typing import Dict, List
 
 import amqp.adapter.file_handler
-from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
+from amqp.model.model import DataMessage
+from amqp.model.snowflake_id import SnowflakeId
 from amqp.adapter.serializer import (
-    map_as_string,
     map_with_list_as_string,
+    map_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,6 +30,7 @@ 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()
@@ -99,10 +100,12 @@ 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(
-            DataMessageMagicByte.HTTP_REQUEST.value,
+            self.MAGIC_BYTE_HTTP_REQUEST,
             self.generate_snowflake_id(),
             method,
             domain,
@@ -119,7 +122,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:
@@ -129,7 +132,9 @@ 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
         )
@@ -144,13 +149,13 @@ class DataMessageFactory:
         """
         return (
             f"DataMessage{{"
-            f"id={message.id()}, "
+            f"id={message.id}, "
-            f"method='{message.method()}', "
+            f"method='{message.method}', "
-            f"domain='{message.domain()}', "
+            f"domain='{message.domain}', "
-            f"path='{message.path()}', "
+            f"path='{message.path}', "
-            f"headers={{{map_with_list_as_string(message.headers())}}}, "
+            f"headers={{{map_with_list_as_string(message.headers)}}}, "
-            f"traceInfo={{{map_as_string(message.trace_info())}}}, "
+            f"traceInfo={{{map_as_string(message.trace_info)}}}, "
-            f"base64body={message.base64body().decode()}"
+            f"base64body={message.base64_body.decode()}"
             f"}}"
         )
 
@@ -161,19 +166,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.base64body())
+        msg_length = 2 + header_length + len(message.base64_body)
         with BytesIO(bytearray(msg_length)) as baos:
             baos.write(prolog_len)
-            baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
+            baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8"))
             baos.write(id_bytes)
             baos.write(prolog)
-            baos.write(message.base64body())
+            baos.write(message.base64_body)
             return baos.getvalue()
 
     @staticmethod
@@ -187,7 +192,7 @@ class DataMessageFactory:
         metadata = input_bytes[2 : prolog_len + 2].decode()
 
         pattern = re.compile(
-            f"{DataMessageMagicByte.HTTP_REQUEST.value}"
+            f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
             r"([0-9A-Fa-f]+)\|"
             r"m=([^|]*)\|"
             r"d=([^|]*)\|"
@@ -215,7 +220,7 @@ class DataMessageFactory:
         trace_info = parse_map_string(trace_info_str)
 
         return DataMessage(
-            DataMessageMagicByte.HTTP_REQUEST.value,
+            DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
             SnowflakeId.from_hex(id_str),
             method,
             domain,

amqp/adapter/data_parser.py

@@ -1,15 +1,15 @@
 import io
 import json
-from typing import Dict, Tuple
+import python_multipart
+
+from fastapi import UploadFile
+from python_multipart.multipart import Field, File
+from typing import Tuple, Dict
 from urllib.parse import parse_qs, urlparse
 from xml.etree import ElementTree
 
-import python_multipart
+from amqp.adapter.logging_utils import logging_error, logging_debug
-from fastapi import UploadFile
+from amqp.model.model import DataMessage
-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,14 +22,15 @@ class MultipartFormDataParser:
     included, the message is transmitted in the unchanged multipart/form-data format.
     """
 
-    def __init__(self, message: AMQMessage):
+    def __init__(self, message: DataMessage):
         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 (
@@ -51,7 +52,9 @@ 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
@@ -67,9 +70,8 @@ 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.decode("utf-8")
+            file.file_object, size=file.size, filename=file.file_name
         )
 
 
@@ -128,11 +130,13 @@ 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: AMQMessage):
+    def parse_request_body(message: DataMessage):
         """
         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.
@@ -142,7 +146,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]
@@ -158,12 +162,15 @@ 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 = AMQDataParser.process_form_data(_form_data)
+                        _form_data = MultipartFormDataParser.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,14 +20,8 @@ class DataResponseFactory:
         :param _id: ID
         :return: Response message
         """
-        return DataResponseFactory.of(
+        return DataResponse(
-            id=_id,
+            str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {}
-            response_code=200,
-            content_type="application/text",
-            body="OK".encode("utf-8"),
-            trace_info={},
-            error="",
-            error_cause="",
         )
 
     @staticmethod
@@ -37,8 +31,6 @@ 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
@@ -50,31 +42,13 @@ class DataResponseFactory:
         :return: DataResponseMessage object
         """
         return DataResponse(
-            magic=DataMessageMagicByte.HTTP_REQUEST.value,
+            id=id.as_string(),
-            id=id,
+            response_code=response_code,
-            response_code=str(response_code),
             content_type=content_type,
-            error=error,
+            response=body,
-            error_cause=error_cause,
+            error=None,
-            headers={},
+            error_cause=None,
             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
@@ -84,19 +58,19 @@ class DataResponseFactory:
         :param response: Object to serialize
         :return: byte array as serialized message
         """
-        id_bytes = response.id().as_string().encode("utf-8")
+        id_bytes = response.id.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.base64body())
+        msg_length = 2 + header_length + len(response.response)
         with BytesIO(bytearray(msg_length)) as baos:
             baos.write(prolog_len)
-            baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
+            baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8"))
             baos.write(id_bytes)
             baos.write(prolog)
-            baos.write(response.base64body())
+            baos.write(base64.b64encode(response.response))
             return baos.getvalue()
 
     @staticmethod
@@ -110,7 +84,7 @@ class DataResponseFactory:
         metadata = input_bytes[2 : prolog_len + 2].decode()
 
         pattern = re.compile(
-            f"{DataMessageMagicByte.HTTP_REQUEST.value}"
+            f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
             r"([0-9A-Fa-f]+)\|"
             r"r=([^|]*)\|"
             r"c=([^|]*)\|"
@@ -136,14 +110,14 @@ class DataResponseFactory:
 
         trace_info = parse_map_string(trace_info_str)
 
-        return DataResponseFactory.of(
+        return DataResponse(
             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 logging
 import os
-from asyncio import AbstractEventLoop
+import logging
+from asyncio import Future, AbstractEventLoop
 from collections import defaultdict
 from pathlib import Path
 
-from aio_pika import Message, connect_robust
+from aio_pika import connect_robust, Message
 from aio_pika.abc import (
-    AbstractExchange,
     AbstractIncomingMessage,
-    AbstractQueue,
     AbstractRobustChannel,
+    AbstractQueue,
+    AbstractExchange,
     DeliveryMode,
 )
 
-from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.router.utils import await_result
 
 # Global dictionary to track completed downloads
@@ -166,6 +166,7 @@ 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"]
@@ -191,7 +192,9 @@ 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():
@@ -234,8 +237,12 @@ class StreamingFileHandler(FileHandler):
                 _res = bytearray()
                 try:
                     channel: AbstractRobustChannel = await _connection.channel()
-                    queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False)
+                    queue: AbstractQueue = await channel.get_queue(
-                    logging_debug(f"Awaiting all buffer chunks being downloaded for {queue_name}")
+                        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():
@@ -274,6 +281,7 @@ 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())
@@ -290,16 +298,22 @@ 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(
@@ -365,7 +379,9 @@ 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
                         ),
                     },
                 )
@@ -378,7 +394,9 @@ 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,11 +4,8 @@ import os
 import sys
 import threading
 import traceback
-from importlib.metadata import version
 
-# Get version from installed package metadata
+verbose = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
-__version__ = version("amq_adapter")  # Name must match pyproject.toml [project] name
-__verbose__ = os.environ.get("AMQ_VERBOSE_LOGGING", "0") == "1"
 
 
 def get_context_info():
@@ -54,18 +51,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}:{__version__} [E] '{msg}' ---> '{_tb.filename}':{_tb.lineno}, '{_tb.name}()'",
+            f"{_thread_id} [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}:{__version__} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+                f"{_thread_id} [E] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
                 *args,
             )
         else:
-            logging.error(f"{_thread_id}:{__version__} [E] {msg}", *args)
+            logging.error(f"{_thread_id} [E] {msg}", *args)
 
 
 def logging_warning(msg: str, *args) -> None:
@@ -76,17 +73,17 @@ def logging_warning(msg: str, *args) -> None:
         msg (str): The warning message to log.
     """
     _thread_id = threading.get_ident()
-    if not __verbose__:
+    if not verbose:
-        logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
+        logging.warning(f"{_thread_id} [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}:{__version__} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [W] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.warning(f"{_thread_id}:{__version__} [W] {msg}", *args)
+        logging.warning(f"{_thread_id} [W] {msg}", *args)
 
 
 def logging_info(msg: str, *args) -> None:
@@ -97,17 +94,17 @@ def logging_info(msg: str, *args) -> None:
         msg (str): The info message to log.
     """
     _thread_id = threading.get_ident()
-    if not __verbose__:
+    if not verbose:
-        logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
+        logging.info(f"{_thread_id} [*] {msg}", *args)
         return
     module, line_number, function_name = get_context_info()
     if module and line_number and function_name:
         logging.info(
-            f"{_thread_id}:{__version__} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [*] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.info(f"{_thread_id}:{__version__} [*] {msg}", *args)
+        logging.info(f"{_thread_id} [*] {msg}", *args)
 
 
 def logging_debug(msg: str, *args) -> None:
@@ -121,8 +118,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}:{__version__} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
+            f"{_thread_id} [DBG] '{msg}' ---> '{module}':{line_number}, '{function_name}()'",
             *args,
         )
     else:
-        logging.debug(f"{_thread_id}:{__version__} [DBG] {msg}", *args)
+        logging.debug(f"{_thread_id} [DBG] {msg}", *args)

amqp/adapter/pydantic_serializer.py

@@ -2,13 +2,12 @@
 This module provides serialization and deserialization functions for Pydantic models that extend BaseModel
 """
 
-import json
 import uuid
+import json
 from datetime import datetime
-from typing import Any, Dict
+from typing import Dict, Any
-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 Dict, List, Union
+from typing import List, Union, Dict
 
 
 def long_to_bytes(x: int) -> bytes:
@@ -63,7 +63,3 @@ 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,12 +1,13 @@
 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"\|"
@@ -98,7 +99,9 @@ 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(
@@ -108,7 +111,11 @@ 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(
@@ -141,7 +148,9 @@ 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

@@ -41,7 +41,9 @@ 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:
@@ -66,9 +68,15 @@ class Dispatch:
 
         :param config: config parser to read configuration values
         """
-        self.use_dlq = config.getboolean("CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False)
+        self.use_dlq = config.getboolean(
-        self.amq_host = config.get("CleverMicro-AMQ", self.PREFIX + "amq-host", fallback="rabbitmq")
+            "CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False
-        self.amq_port = config.getint("CleverMicro-AMQ", self.PREFIX + "amq-port", fallback=5672)
+        )
+        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,41 +2,30 @@
 # CleverMicro AMQ Adapter settings
 # ====================================================================
 [CleverMicro-AMQ]
-# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters
+cm.amq-adapter.service-name=amq-adapter-python
-
+cm.amq-adapter.generator-id=164
-# A name to identify this service for the AMQ Adapter
+cm.amq-adapter.route-mapping=amq-cleverswarm::amq-cleverswarm:cleverswarm.localhost.#:5:0
-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 CleveSwarm nor CleverBRAG or in tight coupling mode.
+# Not used for CleverSwarm 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=cleverthis-service-name.localhost
+#cm.dispatch.service-host=cleverswarm.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
+# Define maximum CPU usage before triggering backpressure OVERLOAD alert
-cm.backpressure.threshold-cpu-overload=90
+cm.backpressure.threshold-cpu=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.cpu-idle-duration=30
+cm.backpressure.idle-duration=30
 

amqp/healthcheck.py

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

amqp/model/model.py

@@ -1,10 +1,8 @@
 import base64
 import json
-from dataclasses import dataclass, field
 from enum import Enum
-from typing import Dict, List
+from typing import List, Dict
-
+from dataclasses import dataclass, field
-from aio_pika.abc import AbstractRobustConnection
 
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
@@ -18,29 +16,9 @@ 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,
@@ -88,19 +66,11 @@ class CleverMicroMessage:
     def body(self) -> bytes:
         return base64.b64decode(self.base64body())
 
-    def content_type(self) -> str:
+    def contentType(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,
@@ -112,7 +82,9 @@ 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()
@@ -123,77 +95,19 @@ 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,
@@ -265,6 +179,65 @@ 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
 
@@ -293,7 +266,7 @@ class ScalingRequestAlert(Enum):
     IDLE = "IDLE"
     UPDATE = "UPDATE"
     SCALE_UP = "SCALE_UP"
-    SCALE_DOWN = "SCALE_DOWN"
+    SCSLE_DOWN = "SCALE_DOWN"
 
 
 @dataclass(frozen=True)
@@ -328,32 +301,7 @@ 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,21 +1,20 @@
 import logging
 
 import aio_pika
-from aio_pika.abc import AbstractRobustExchange, AbstractRobustQueue
+from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange, ExchangeType
-from pika.exchange_type import ExchangeType
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import (
-    logging_debug,
     logging_error,
+    logging_debug,
     logging_info,
     logging_warning,
 )
-from amqp.model.model import AMQRoute, DataMessage
+from amqp.model.model import DataMessage, AMQRoute
 from amqp.router.router_base import (
     AMQ_REPLY_TO_EXCHANGE,
-    AMQ_RPC_EXCHANGE,
     DLQ_EXCHANGE,
+    AMQ_RPC_EXCHANGE,
 )
 from amqp.router.router_consumer import RouterConsumer
 from amqp.router.utils import sanitize_as_rabbitmq_name
@@ -55,7 +54,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
@@ -85,15 +84,21 @@ 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
+            queue_args = (
-            logging_info("RabbitMQClient register routes, cfg mapping: [%s]", route_mapping)
+                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(
@@ -103,9 +108,13 @@ class RabbitMQClient:
                         auto_delete=False,
                         arguments=queue_args,
                     )
-                    _exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE
+                    _exchange_name: str = (
-                    exchange: AbstractRobustExchange = await self.channel.declare_exchange(
+                        route.exchange if route.exchange else AMQ_RPC_EXCHANGE
-                        name=_exchange_name, type=ExchangeType.topic
+                    )
+                    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)
@@ -190,14 +199,16 @@ 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(
@@ -208,14 +219,12 @@ 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 not sent, no route for %s/%s", message.domain, message.path
-                    message.domain(),
-                    message.path(),
                 )
 
         except Exception as err:

amqp/router/route_database.py

@@ -1,3 +1,4 @@
+import logging
 import re
 from typing import Set
 
@@ -40,9 +41,13 @@ 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,7 +1,8 @@
+import logging
 from typing import Callable, List, Set
 
-from amqp.adapter.amq_route_factory import NULL_ROUTE, AMQRouteFactory
+from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE
-from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.logging_utils import logging_error, logging_debug, 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
@@ -27,7 +28,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()
@@ -37,7 +38,10 @@ 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]:
@@ -47,12 +51,14 @@ 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:
@@ -76,9 +82,14 @@ 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 AMQRoute, ServiceMessage
+from amqp.model.model import ServiceMessage, AMQRoute
 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,13 +77,17 @@ 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__()
@@ -97,7 +101,9 @@ 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]",
@@ -118,7 +124,9 @@ 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",
@@ -150,7 +158,9 @@ 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:
@@ -172,7 +182,9 @@ 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,
@@ -189,9 +201,13 @@ 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,20 +7,21 @@
 * forwards it to the application.
 """
 
+import logging
+
 from aio_pika import Message
 from aio_pika.abc import (
-    AbstractIncomingMessage,
     AbstractRobustChannel,
+    AbstractRobustQueue,
     AbstractRobustConnection,
     AbstractRobustExchange,
-    AbstractRobustQueue,
+    AbstractIncomingMessage, ExchangeType,
 )
-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
@@ -28,13 +29,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 (
-    ANY_RECIPIENT,
+    RouterBase,
-    CM_C_AMQ_QUEUE,
-    CM_P_REPLY_TO,
-    CM_P_RESP_QUEUE,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
-    RouterBase,
+    CM_C_AMQ_QUEUE,
+    CM_P_RESP_QUEUE,
+    CM_P_REPLY_TO,
+    ANY_RECIPIENT,
 )
 
 
@@ -47,14 +48,16 @@ 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(
-            "RouterProducer.<init>: %s, route: %s",
+            f"RouterProducer.<init>: %s, route: %s",
             self.amq_configuration.amq_adapter.service_name,
             self.amq_configuration.amq_adapter.route_mapping,
         )
@@ -78,13 +81,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()
@@ -96,7 +99,9 @@ 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
@@ -110,7 +115,9 @@ 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):
         """
@@ -118,13 +125,15 @@ class RouterProducer(RouterBase):
         * routing data, invokes addRoute or removeRoute depending on the type of the message.
         """
         logging_info(
-            "[%s] Received ROUTING DATA: [%s], msgId=%s",
+            f"[%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",
@@ -135,8 +144,9 @@ class RouterProducer(RouterBase):
             cm_message.reply_to,
         )
         if (
-            cm_message.is_valid()
+                cm_message.is_valid()
-            and not self.amq_configuration.amq_adapter.service_name == cm_message.reply_to
+                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)
@@ -155,7 +165,9 @@ 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,
@@ -165,7 +177,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
-from asyncio import AbstractEventLoop, Future
+import os
+from asyncio import Future, AbstractEventLoop
 from typing import Dict
 
 from aio_pika import Message
 from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
+
 from opentelemetry import trace
-from opentelemetry.trace import Tracer, TraceState
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+from opentelemetry.trace import Tracer, TraceState
 
 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.file_handler import FileHandler, StreamingFileHandler
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
-from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
-from amqp.adapter.serializer import map_as_string
+from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import (
-    DataMessage,
-    DataMessageMagicByte,
     DataResponse,
+    DataMessage,
     MultipartDataMessage,
 )
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
+from amqp.adapter.serializer import map_as_string
 
 
 def collect_trace_info():
@@ -75,15 +75,12 @@ class DataMessageHandler:
         self.file_handler: FileHandler = file_handler
         self.backpressure_handler: BackpressureHandler = backpressure_handler
 
-    async def inbound_data_message_callback_as_thread(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback(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.inbound_data_message_callback_no_thread(message))
+            target=lambda: asyncio.run(self.run_callback_thread(message))
         ).start()
 
-    async def inbound_data_message_callback_no_thread(self, message: AbstractIncomingMessage):
+    async def run_callback_thread(self, message: AbstractIncomingMessage):
         """
         Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
         Ensures Jaeger trace-id propagation.
@@ -95,17 +92,19 @@ 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() == DataMessageMagicByte.HTTP_REQUEST.value:
+            if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST:
                 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)
 
@@ -169,7 +168,9 @@ 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,
@@ -178,24 +179,26 @@ 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.id,
-                amq_message.method(),
+                amq_message.method,
-                amq_message.path(),
+                amq_message.path,
-                map_as_string(amq_message.trace_info()),
+                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)
@@ -204,7 +207,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
@@ -226,15 +229,17 @@ 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]
+                    response.content_type[0]
-                    if isinstance(response.content_type(), list)
+                    if isinstance(response.content_type, list)
-                    else response.content_type()
+                    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(
@@ -252,12 +257,15 @@ 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,24 +1,26 @@
-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")
@@ -64,7 +66,9 @@ 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,
@@ -103,7 +107,7 @@ class AMQService:
             try:
                 await self.rabbit_mq_client.register_inbound_routes(
                     route_mapping,
-                    self.amq_data_message_handler.inbound_data_message_callback_no_thread,
+                    self.amq_data_message_handler.inbound_data_message_callback,
                 )
                 return await self.rabbit_mq_client.register_data_reply_callback(
                     self.amq_data_message_handler.reply_received_callback
@@ -118,7 +122,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,13 +1,14 @@
-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 TracerProvider, Tracer
-from opentelemetry.sdk.trace import Tracer, TracerProvider
 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
 
 
 def initialize_trace(app=None, name=None) -> Tracer:
@@ -16,7 +17,9 @@ 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

@@ -1,122 +0,0 @@
-"""
-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

@@ -1,41 +0,0 @@
-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,51 +5,129 @@ 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 Any, List, Optional
+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
 
 # CleverSwarm specific imports
 import core.deps
-from fastapi.routing import APIRoute
+from rest.v0 import api
 from schemas.user_schema import UserSchema
 
-from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
+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.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]:
     """
-    Get the function/Callable for the module_name.function_name if it exists in the module, else return None.
+    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.
     """
+
+    # 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)
-        _preferred_endpoint = getattr(_module, function_name)
+        _prefered_endpoint = getattr(_module, function_name)
-        return _preferred_endpoint if callable(_preferred_endpoint) else None
+        return _prefered_endpoint if callable(_prefered_endpoint) else None
     except Exception:
         return None
 
 
-# =================================================================================================
+def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]:
-# ======================== C l e v e r S w a r m   A M Q    A d a p t e r  ========================
+    """
-# =================================================================================================
+    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)
+
+
 class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
     """
-    CleverSwarm AMQP Adapter for CleverSwarm API. Implements the CleverSwarm specific logic that overrides
+    CleverSwarm AMQP Adapter for CleverSwarm API. Implements the interface to CleverSwarm API.
-    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, routes: List[APIRoute]):
+    def __init__(
-        super().__init__(amq_configuration, routes=routes, session=None)
+        self, amq_configuration: AMQConfiguration, session: ClientSession = None
+    ):
+        super().__init__(amq_configuration, session)
+        self.routes = group_routes_by_method(api_routes=api.api_router.routes)
         _amq_service: AMQService = AMQService(amq_configuration, self)
         self.thread = Thread(target=_amq_service.run)
         self.thread.start()
@@ -57,25 +135,168 @@ 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)
 
-    def get_endpoint(self, endpoint: Any) -> Any:
+    # ==================================================================================================
-        """
+    # =====================  C l e v e r S w a r m   A P I   m e t h o d s  ============================
-        If the specific endpoint has a matching JSON wrapper (identified as {endpoint.name}_json ),
+    # ==================================================================================================
-        then use it as preferred way to invoke the endpoint.
 
-        :param endpoint: the Callable for the endpoint
+    async def get(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
-
-        :return: new Callable representing the JSON wrapper endpoint, if it exists or the original Callable.
         """
-        _preferred_endpoint = get_callable_for_name(
+        Handles the GET requests for the CleverSwarm API.
-            getattr(endpoint, "__module__"),
+        :param message: message from the AMQP that contains the GET request
-            endpoint.__name__ + PREFERRED_SUFFIX,
+        :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")
+
+    async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
+        """
+        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,
         )
-        return _preferred_endpoint if _preferred_endpoint is not None else endpoint

cleverswarm/main.py

@@ -1,7 +1,7 @@
-from clever_swarm_adapter import CleverSwarmAmqpAdapter
-
 from amqp.config.amq_configuration import AMQConfiguration
 
+from clever_swarm_adapter import CleverSwarmAmqpAdapter
+
 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.31"
+version = "0.2.23-dev"
-description = "The Python implementation of the AMQ Adapter for CleverMicro"
+description = "The Python implementation of the AMQ Adaptor for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -19,16 +19,15 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika",
+    "aio-pika~=9.5",
-    "aiohttp",
+    "aiohttp~=3.11",
-    "fastapi",
+    "aio_pika~=9.5",
+    "fastapi~=0.114",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
-    "pika",
-    "psutil",
     "python_multipart"
 ]
 
@@ -36,18 +35,11 @@ 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 AMQMessage, CleverThisServiceAdapter
+from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
 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, routes=[], session=self.mock_session
+            amq_configuration=self.mock_amq_config, 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,7 +91,9 @@ 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()
@@ -137,7 +139,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"},
@@ -153,7 +155,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)
@@ -172,12 +174,12 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "_authorize_forward_request",
+            "handle_possible_form",
-            wraps=self.adapter._authorize_forward_request,
+            wraps=self.adapter.handle_possible_form,
         ) 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):
@@ -197,12 +199,12 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "_authorize_forward_request",
+            "handle_possible_form",
-            wraps=self.adapter._authorize_forward_request,
+            wraps=self.adapter.handle_possible_form,
         ) 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,21 +10,25 @@ 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()
@@ -61,9 +65,9 @@ class DataMessageFactoryTest(TestCase):
             {"k1": ["v1"], "k2": ["v2"]},
             "Quick brown fox",
         )
-        self.assertEqual("localhost", _req.domain(), "Domain 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("/test/me", _req.path, "Context path not set correctly")
-        self.assertEqual("v1", _req.headers()["k1"][0], "Headers 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):
@@ -78,11 +82,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(_req.id.as_string(), _deser.id.as_string())
-        self.assertEqual("localhost", _deser.domain())
+        self.assertEqual("localhost", _deser.domain)
-        self.assertEqual("/test/me", _deser.path())
+        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)
@@ -120,4 +124,6 @@ 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,7 +2,6 @@ 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
 
 
@@ -12,8 +11,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(200, _resp.response_code)
-        self.assertEqual(b"OK", _resp.body())
+        self.assertEqual(b"OK", _resp.response)
 
     def test_serialize(self):
         _f = DataResponseFactory()
@@ -23,7 +22,7 @@ class TestDataResponseFactory(TestCase):
         _ser = _f.serialize(_resp)
         self.assertEqual(
             2,
-            str(_ser[2:]).index(DataMessageMagicByte.HTTP_REQUEST.value + _snowflakeId.as_string()),
+            str(_ser[2:]).index(_f.MAGIC_BYTE_HTTP_REQUEST + _snowflakeId.as_string()),
             "Magic byte and SnowflakeId not found",
         )
         self.assertGreater(str(_ser[2:]).index("|e=|"), 2)
@@ -36,6 +35,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(_snowflakeId.as_string(), _deser.id)
-        self.assertEqual("200", _deser.response_code())
+        self.assertEqual(200, _deser.response_code)
-        self.assertEqual(b"OK", _deser.body())
+        self.assertEqual(b"OK", _deser.response)

tests/router/test_route_database.py

@@ -2,8 +2,8 @@ from unittest import TestCase
 
 import pytest
 
-from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.data_message_factory import DataMessageFactory
+from amqp.adapter.amq_route_factory import AMQRouteFactory
 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

@@ -1,90 +0,0 @@
-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()