#43 - fix the issues reported by flake8

Reviewed-on: #42

.flake8

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

.pre-commit-config.yaml

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

amqp/adapter/backpressure_handler.py

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

amqp/adapter/cleverthis_service_adapter.py

@@ -7,24 +7,25 @@ import asyncio
 import inspect
 import json
 from asyncio import AbstractEventLoop
-from typing import Dict, List, Callable, Coroutine, Any
+from dataclasses import dataclass
+from typing import Any, Callable, Coroutine, Dict, List
 
 from aio_pika.abc import (
+    AbstractExchange,
     AbstractRobustChannel,
     AbstractRobustConnection,
-    AbstractExchange,
 )
-from aiohttp import ClientSession, ClientResponse
-from dataclasses import dataclass
+from aiohttp import ClientResponse, ClientSession
 from fastapi import HTTPException
+from pydantic import BaseModel
 from starlette.responses import FileResponse
 
-from amqp.adapter.data_parser import AMQDataParser
+from amqp.adapter import serializer
 from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import DataMessage, DataResponse
+from amqp.model.model import AMQErrorMessage, DataMessage, DataResponse
 from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
 from amqp.router.utils import await_result
 
@@ -95,7 +96,7 @@ async def _convert_file_response(
         _exchange, obj.path, _routing_key, loop
     )
     # as above, but luckily this time we don't need to wait for the result
-    _future = asyncio.run_coroutine_threadsafe(_channel.close(), loop)
+    asyncio.run_coroutine_threadsafe(_channel.close(), loop)
     return json.dumps(
         {
             "files": [
@@ -208,8 +209,10 @@ class CleverThisServiceAdapter:
             else:
                 raise ValueError(f"Unexpected HTTP method: {message.method}")
         else:
-            _amq_response = DataResponseFactory.of(
-                message.id, 401, "text/plain", b"Unauthorized", message.trace_info
+            _amq_response = DataResponseFactory.of_error_message(
+                message.id,
+                AMQErrorMessage(401, "Unauthorized", "Unauthorized"),
+                message.trace_info,
             )
         logging_info(
             f"ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}"
@@ -244,11 +247,9 @@ class CleverThisServiceAdapter:
                 await _http_resp.read(),
                 message.trace_info,
             )
-        return DataResponseFactory.of(
+        return DataResponseFactory.of_error_message(
             message.id,
-            404,
-            "text/plain",
-            str("Destination location does not exists").encode("utf-8"),
+            AMQErrorMessage(404, "Not Found", "Destination location does not exists"),
             message.trace_info,
         )
 
@@ -260,7 +261,7 @@ class CleverThisServiceAdapter:
         :param auth_user: user authenticated by the API Gateway, as UserSchem
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        return await self.handle_possible_form(
+        return await self._authorize_forward_request(
             message,
             self.session.put(
                 f"http://{self.service_host}:{self.service_port}{message.path}",
@@ -278,7 +279,7 @@ class CleverThisServiceAdapter:
         :param auth_user: user authenticated by the API Gateway, as UserSchem
         :return: DataResponse to be sent back to the API Gateway via AMQP
         """
-        return await self.handle_possible_form(
+        return await self._authorize_forward_request(
             message,
             self.session.post(
                 f"http://{self.service_host}:{self.service_port}{message.path}",
@@ -288,10 +289,14 @@ class CleverThisServiceAdapter:
             auth_user,
         )
 
-    async def handle_possible_form(
+    async def _authorize_forward_request(
         self, message: DataMessage, request_coroutine, auth_user: Any
     ) -> DataResponse:
-        _args = AMQDataParser.parse_request_body(message)
+        """
+        This method is used to authorize the request and forward it to the CleverThis Service API.
+        However, it is only used for loose / 3rd party mode of coupling with the Service
+        """
+        # AMQDataParser.parse_request_body(message)
         return await request_coroutine
 
     async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
@@ -345,7 +350,9 @@ class CleverThisServiceAdapter:
                     elif arg.annotation == float:
                         _verified_args[arg.name] = float(args[arg.name])
                     elif arg.annotation == bool:
-                        _verified_args[arg.name] = bool(args[arg.name])
+                        _verified_args[arg.name] = serializer.str_to_bool(
+                            args[arg.name]
+                        )
                     # TODO: more?
                     else:
                         _verified_args[arg.name] = args[arg.name]
@@ -371,6 +378,9 @@ class CleverThisServiceAdapter:
                     ),
                     "application/octet-stream",
                 )
+            elif isinstance(_result, BaseModel):
+                _json_result = _result.json()
+                _content_type = "application/json"
             else:
                 _json_result = _result
                 _content_type = "application/json"
@@ -399,10 +409,18 @@ class CleverThisServiceAdapter:
                 message.trace_info,
             )
         except HTTPException as http_error:
+            logging_error(f"Service REST endpoint invocation failed: {http_error}")
             _content = str(http_error.detail)
             if not is_likely_json(_content):
-                _content = '{"error": "' + _content + '"}'
-            logging_error(f"Service REST endpoint invocation failed: {http_error}")
+                return DataResponseFactory.of_error_message(
+                    message.id,
+                    AMQErrorMessage(
+                        http_error.status_code,
+                        "Service Invocation Failed",
+                        _content,
+                    ),
+                    message.trace_info,
+                )
             return DataResponseFactory.of(
                 message.id,
                 http_error.status_code,
@@ -414,7 +432,15 @@ class CleverThisServiceAdapter:
             logging_error(f"Service endpoint invocation failed: {error}")
             _content = str(error)
             if not is_likely_json(_content):
-                _content = '{"error": "' + _content + '"}'
+                return DataResponseFactory.of_error_message(
+                    message.id,
+                    AMQErrorMessage(
+                        500,
+                        "Service Invocation Failed",
+                        _content,
+                    ),
+                    message.trace_info,
+                )
             return DataResponseFactory.of(
                 message.id,
                 500,

amqp/adapter/data_parser.py

@@ -1,6 +1,7 @@
 import io
 import json
 import python_multipart
+from tempfile import SpooledTemporaryFile
 
 from fastapi import UploadFile
 from python_multipart.multipart import Field, File
@@ -70,8 +71,12 @@ class MultipartFormDataParser:
 
     def on_file(self, file: File):
         logging_debug(str(file))
+        file.file_object.seek(0)
+        _spooled_file: SpooledTemporaryFile = SpooledTemporaryFile()
+        _spooled_file.write(file.file_object.read())
+        _spooled_file.seek(0)
         self.form_data[file.field_name.decode("utf-8")] = UploadFile(
-            file.file_object, size=file.size, filename=file.file_name
+            _spooled_file, size=file.size, filename=file.file_name.decode("utf-8")
         )
 
 

amqp/adapter/data_response_factory.py

@@ -5,7 +5,7 @@ method to create, serialize and deserialize DataResponse message
 import base64
 import re
 from io import BytesIO
-from amqp.model.model import DataResponse
+from amqp.model.model import DataResponse, AMQErrorMessage
 from amqp.model.snowflake_id import SnowflakeId
 from amqp.adapter.serializer import parse_map_string, map_as_string
 
@@ -31,6 +31,8 @@ class DataResponseFactory:
         content_type: str,
         body: bytes,
         trace_info: dict[str, str],
+        error: str | None = None,
+        error_cause: str | None = None,
     ) -> DataResponse:
         """
         Create the DataResponse message supplying all key values
@@ -46,11 +48,27 @@ class DataResponseFactory:
             response_code=response_code,
             content_type=content_type,
             response=body,
-            error=None,
-            error_cause=None,
+            error=error,
+            error_cause=error_cause,
             trace_info=trace_info,
         )
 
+    @staticmethod
+    def of_error_message(
+        id: SnowflakeId,
+        error_message: AMQErrorMessage,
+        trace_info: dict[str, str],
+    ) -> DataResponse:
+        return DataResponseFactory.of(
+            id,
+            error_message.response_code,
+            error_message.content_type,
+            error_message.serialize(),
+            trace_info,
+            error=error_message.error,
+            error_cause=error_message.detail,
+        )
+
     @staticmethod
     def serialize(response: DataResponse) -> bytes:
         """

amqp/adapter/file_handler.py

@@ -4,22 +4,22 @@ implements the file downloader for the amqp adapter, where file is sent in chunk
 
 import asyncio
 import json
-import os
 import logging
-from asyncio import Future, AbstractEventLoop
+import os
+from asyncio import AbstractEventLoop
 from collections import defaultdict
 from pathlib import Path
 
-from aio_pika import connect_robust, Message
+from aio_pika import Message, connect_robust
 from aio_pika.abc import (
-    AbstractIncomingMessage,
-    AbstractRobustChannel,
-    AbstractQueue,
     AbstractExchange,
+    AbstractIncomingMessage,
+    AbstractQueue,
+    AbstractRobustChannel,
     DeliveryMode,
 )
 
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
 from amqp.router.utils import await_result
 
 # Global dictionary to track completed downloads
@@ -166,7 +166,6 @@ class StreamingFileHandler(FileHandler):
             message_data (dict): The entire message
             output_dir (str): The directory where the file should be saved.
         """
-        global download_locations
         filename = file_info["filename"]
         size = file_info["size"]
         stream_name = file_info["streamName"]
@@ -281,7 +280,6 @@ class StreamingFileHandler(FileHandler):
             loop: The asyncio event loop of the RabbitMQ API
             output_dir
         """
-        global download_locations
         try:
             _message_data = json.loads(message.body.decode())
             _amq_message_data = json.loads(json_data.decode())

amqp/adapter/logging_utils.py

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

amqp/adapter/serializer.py

@@ -63,3 +63,7 @@ def parse_map_list_string(input_str: str) -> Dict[str, List[str]]:
         for token in input_str[1:-1].split("],"):
             add_to_map(map_, token)
     return map_
+
+
+def str_to_bool(value: str) -> bool:
+    return value.lower() in ("true", "yes", "1", "t", "y", "on")

amqp/adapter/service_message_factory.py

@@ -1,13 +1,12 @@
 import base64
-import logging
 
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.logging_utils import logging_error, logging_info
+from amqp.adapter.serializer import map_as_string, parse_map_string
 from amqp.adapter.trace_info_adapter import TraceInfoAdapter
 from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
-from amqp.adapter.serializer import parse_map_string, map_as_string
 
 RS = "|"
 SPLIT_RS = r"\|"

amqp/config/amq_configuration.py

@@ -5,7 +5,7 @@ import os
 from amqp.adapter.logging_utils import logging_error, logging_warning
 
 """
-Convert configuration from properties file to an object. 
+Convert configuration from properties file to an object.
 """
 
 

amqp/config/application.properties

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

amqp/model/model.py

@@ -1,8 +1,8 @@
 import base64
 import json
-from enum import Enum
-from typing import List, Dict
 from dataclasses import dataclass, field
+from enum import Enum
+from typing import Dict, List
 
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.model.snowflake_id import SnowflakeId
@@ -16,9 +16,24 @@ Define the structure of messages used in Clever Micro
 """
 
 
+@dataclass
 class CleverMicroMessage:
     DEFAULT_CONTENT_TYPE = ["application/octet-stream"]
 
+    """
+    The common structure of a message that transfers the user request to the Service and back. All data related messages
+    conform to this base structure:
+
+    magic - identifies the message type (REST, RPC, DataResponse)
+    id - unique message ID
+    m_field - method name (for REST style calls) or service name (for RPC style calls) or response code for DataResponse
+    d_field - domain name (for REST style calls) or module name (for RPC style calls) or  error code for DataResponse
+    p_field - path (for REST style calls) or method name (for RPC style calls) or error cause for DataResponse
+    headers - headers of the message (e.g. Content-Type, Accept, etc.)
+    trace_info - trace information (e.g. trace ID, span ID, etc. for Jaeger trace monitor)
+    base64body - the body of the message encoded in base64
+    """
+
     def __init__(
         self,
         magic: str,
@@ -66,11 +81,19 @@ class CleverMicroMessage:
     def body(self) -> bytes:
         return base64.b64decode(self.base64body())
 
-    def contentType(self) -> str:
+    def content_type(self) -> str:
         return self.headers().get("Content-Type", self.DEFAULT_CONTENT_TYPE)[0]
 
 
 class DataMessage(CleverMicroMessage):
+    """
+    The class represents a structure of a message that transfers the user request to the Service.
+    It may also be used to make a call/request between services.
+    The response to this call is in format of DataResponse class.
+
+    See DataMessageFactory for code to instantiate, serialize and deserialize the class
+    """
+
     def __init__(
         self,
         magic: str,
@@ -95,8 +118,17 @@ 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 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,
@@ -179,31 +211,6 @@ class AMQRoute:
         return self.__str__()
 
 
-@dataclass
-class DataMessage:
-    """
-    The class represents a structure of a message that transfers the user request to the Service.
-    It may also be used to make a call/request between services.
-    The response to this call is in format of DataResponse class.
-    Please see DataMessageFactory for code to instantiate, serialize and deserialize the class
-    """
-
-    magic: str
-    id: SnowflakeId
-    method: str
-    domain: str
-    path: str
-    headers: dict[str, List[str]]
-    trace_info: dict[str, str]
-    base64_body: bytes
-
-    def routing_key(self) -> str:
-        return sanitize_as_rabbitmq_name(f"{self.domain}.{self.path}")
-
-    def body(self) -> bytes:
-        return base64.b64decode(self.base64_body)
-
-
 class MultipartDataMessage(DataMessage):
     def __init__(self, message: DataMessage, extra_data: dict):
         super().__init__(
@@ -219,25 +226,6 @@ class MultipartDataMessage(DataMessage):
         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
 
@@ -305,3 +293,30 @@ class ScalingRequest:
                 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/router/route_database.py

@@ -1,4 +1,3 @@
-import logging
 import re
 from typing import Set
 

amqp/router/router_base.py

@@ -1,8 +1,7 @@
-import logging
 from typing import Callable, List, Set
 
-from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.amq_route_factory import NULL_ROUTE, AMQRouteFactory
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
 from amqp.model.model import AMQRoute, DataMessage
 from amqp.router.route_database import RouteDatabase
 from amqp.router.utils import sanitize_as_rabbitmq_name

amqp/router/router_consumer.py

@@ -1,26 +1,26 @@
-import logging
 from typing import Set
 
 from aio_pika.abc import (
+    AbstractIncomingMessage,
     AbstractRobustChannel,
     AbstractRobustQueue,
-    AbstractIncomingMessage,
 )
+
 from amqp.adapter.amq_route_factory import AMQRouteFactory
 from amqp.adapter.logging_utils import (
+    logging_debug,
     logging_error,
     logging_info,
-    logging_debug,
     logging_warning,
 )
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import ServiceMessage, AMQRoute
+from amqp.model.model import AMQRoute, ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import (
+    ANY_RECIPIENT,
     CM_C_REQ_QUEUE,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
-    ANY_RECIPIENT,
 )
 from amqp.router.router_producer import RouterProducer
 

amqp/router/router_producer.py

@@ -7,22 +7,20 @@
 * forwards it to the application.
 """
 
-import logging
-
 from aio_pika import Message
 from aio_pika.abc import (
+    AbstractIncomingMessage,
     AbstractRobustChannel,
-    AbstractRobustQueue,
     AbstractRobustConnection,
     AbstractRobustExchange,
-    AbstractIncomingMessage,
+    AbstractRobustQueue,
 )
 from pika.exchange_type import ExchangeType
 
 from amqp.adapter.logging_utils import (
+    logging_debug,
     logging_error,
     logging_info,
-    logging_debug,
     logging_warning,
 )
 from amqp.adapter.service_message_factory import ServiceMessageFactory
@@ -30,13 +28,13 @@ from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import ServiceMessage
 from amqp.model.service_message_type import ServiceMessageType
 from amqp.router.router_base import (
-    RouterBase,
+    ANY_RECIPIENT,
+    CM_C_AMQ_QUEUE,
+    CM_P_REPLY_TO,
+    CM_P_RESP_QUEUE,
     CM_REQUEST_EXCHANGE,
     CM_RESPONSE_EXCHANGE,
-    CM_C_AMQ_QUEUE,
-    CM_P_RESP_QUEUE,
-    CM_P_REPLY_TO,
-    ANY_RECIPIENT,
+    RouterBase,
 )
 
 
@@ -58,7 +56,7 @@ class RouterProducer(RouterBase):
         self.cm_response_exchange: AbstractRobustExchange | None = None
 
         logging_info(
-            f"RouterProducer.<init>: %s, route: %s",
+            "RouterProducer.<init>: %s, route: %s",
             self.amq_configuration.amq_adapter.service_name,
             self.amq_configuration.amq_adapter.route_mapping,
         )
@@ -126,7 +124,7 @@ class RouterProducer(RouterBase):
         * routing data, invokes addRoute or removeRoute depending on the type of the message.
         """
         logging_info(
-            f"[%s] Received ROUTING DATA: [%s], msgId=%s",
+            "[%s] Received ROUTING DATA: [%s], msgId=%s",
             message.delivery_tag,
             message.body,
             message.message_id,

amqp/service/amq_message_handler.py

@@ -1,32 +1,31 @@
 import asyncio
+import os
 import threading
 import time
-import os
-from asyncio import Future, AbstractEventLoop
+from asyncio import AbstractEventLoop, Future
 from typing import Dict
 
 from aio_pika import Message
 from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange
-
 from opentelemetry import trace
-from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 from opentelemetry.trace import Tracer, TraceState
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 
 from amqp.adapter.backpressure_handler import BackpressureHandler
+from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
 from amqp.adapter.data_message_factory import DataMessageFactory
 from amqp.adapter.data_response_factory import DataResponseFactory
-from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
-from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
-from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
+from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
+from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info
+from amqp.adapter.serializer import map_as_string
 from amqp.adapter.service_message_factory import ServiceMessageFactory
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import (
-    DataResponse,
     DataMessage,
+    DataResponse,
     MultipartDataMessage,
 )
 from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient
-from amqp.adapter.serializer import map_as_string
 
 
 def collect_trace_info():
@@ -75,12 +74,21 @@ class DataMessageHandler:
         self.file_handler: FileHandler = file_handler
         self.backpressure_handler: BackpressureHandler = backpressure_handler
 
-    async def inbound_data_message_callback(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback_as_thread(
+        self, message: AbstractIncomingMessage
+    ):
+        """
+        The callback method that is called when a DataMessage is received from RabbitMQ, starts new Thread.
+        """
         threading.Thread(
-            target=lambda: asyncio.run(self.run_callback_thread(message))
+            target=lambda: asyncio.run(
+                self.inbound_data_message_callback_no_thread(message)
+            )
         ).start()
 
-    async def run_callback_thread(self, message: AbstractIncomingMessage):
+    async def inbound_data_message_callback_no_thread(
+        self, message: AbstractIncomingMessage
+    ):
         """
         Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object
         Ensures Jaeger trace-id propagation.
@@ -257,7 +265,6 @@ 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)

amqp/service/amq_service.py

@@ -107,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,
+                    self.amq_data_message_handler.inbound_data_message_callback_no_thread,
                 )
                 return await self.rabbit_mq_client.register_data_reply_callback(
                     self.amq_data_message_handler.reply_received_callback

cleverswarm/clever_swarm_adapter.py

@@ -26,7 +26,7 @@ from amqp.adapter.data_parser import AMQDataParser
 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.model.model import AMQResponse, AMQErrorMessage
 from amqp.service.amq_service import AMQService
 
 
@@ -220,12 +220,12 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                     loop=self.loop,
                 )
 
-        return AMQResponseFactory.of(
+        return AMQResponseFactory.of_error_message(
             message.id,
-            404,
-            "text/plain",
-            str(f"Destination location [{message.path}] does not exists").encode(
-                "utf-8"
+            AMQErrorMessage(
+                404,
+                "Not Found",
+                f"Destination location [{message.path}] does not exists",
             ),
             message.trace_info,
         )
@@ -252,11 +252,13 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                 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 message.extra_data.get(_body_param.name)
                             and UploadFile(
-                                file=pathlib.Path(actual_filepath).open("rb"),
+                                file=pathlib.Path(
+                                    message.extra_data.get(_body_param.name)
+                                ).open("rb"),
                                 size=_file_data["size"],
                                 filename=_file_data["filename"],
                                 headers=Headers(
@@ -291,12 +293,12 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
                     loop=self.loop,
                 )
 
-        return AMQResponseFactory.of(
+        return AMQResponseFactory.of_error_message(
             message.id,
-            404,
-            "text/plain",
-            str(f"Destination location [{message.path}] does not exists").encode(
-                "utf-8"
+            AMQErrorMessage(
+                404,
+                "Not Found",
+                f"Destination location [{message.path}] does not exists",
             ),
             message.trace_info,
         )

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.23-dev"
-description = "The Python implementation of the AMQ Adaptor for CleverMicro"
+version = "0.2.30"
+description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [
     { name = "Stanislav Hejny", email = "stanislav.hejny@cleverthis.com" }
@@ -19,15 +19,16 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika~=9.5.5",
-    "aiohttp~=3.11.11",
-    "aio_pika~=9.5.5",
-    "fastapi==0.114.2",
+    "aio-pika",
+    "aiohttp",
+    "fastapi",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
+    "pika",
+    "psutil",
     "python_multipart"
 ]
 

tests/adapter/test_cleverthis_service_adapter.py

@@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 from aiohttp import ClientResponse
 
-from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
+from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter
 from amqp.config.amq_configuration import AMQConfiguration
 from amqp.model.model import DataMessage, DataResponse
 
@@ -174,8 +174,8 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "handle_possible_form",
-            wraps=self.adapter.handle_possible_form,
+            "_authorize_forward_request",
+            wraps=self.adapter._authorize_forward_request,
         ) as mock_handler:
             response = await self.adapter.put(mock_message, None)
 
@@ -199,8 +199,8 @@ class TestCleverThisServiceAdapter(unittest.TestCase):
 
         with patch.object(
             self.adapter,
-            "handle_possible_form",
-            wraps=self.adapter.handle_possible_form,
+            "_authorize_forward_request",
+            wraps=self.adapter._authorize_forward_request,
         ) as mock_handler:
             response = await self.adapter.post(mock_message, None)
 

tools/create_exchange.py

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