#40 - ensure the JSON format is handled correctly at all levels

Reviewed-on: #39

amqp/adapter/backpressure_handler.py

@@ -178,15 +178,15 @@ class BackpressureHandler:
                 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,
-                )
+                # 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

amqp/adapter/cleverthis_service_adapter.py

@@ -17,14 +17,16 @@ from aio_pika.abc import (
 from aiohttp import ClientSession, ClientResponse
 from dataclasses import dataclass
 from fastapi import HTTPException
+from pydantic import BaseModel
 from starlette.responses import FileResponse
 
+from amqp.adapter import serializer, pydantic_serializer
 from amqp.adapter.data_parser import AMQDataParser
 from amqp.adapter.data_response_factory import DataResponseFactory
 from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
 from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
 from amqp.config.amq_configuration import AMQConfiguration
-from amqp.model.model import DataMessage, DataResponse
+from amqp.model.model import DataMessage, DataResponse, AMQErrorMessage
 from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
 from amqp.router.utils import await_result
 
@@ -208,8 +210,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 +248,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,
         )
 
@@ -345,7 +347,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 +375,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 +406,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 +429,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/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/model/model.py

@@ -305,3 +305,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")

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,
         )
@@ -293,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,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.27"
+version = "0.2.30"
 description = "The Python implementation of the AMQ Adapter for CleverMicro"
 readme = "README.md"
 authors = [
@@ -19,15 +19,15 @@ classifiers = [
 ]
 requires-python = ">=3.11"
 dependencies = [
-    "aio-pika~=9.5.5",
-    "aiohttp~=3.11.11",
-    "fastapi==0.114.2",
+    "aio-pika",
+    "aiohttp",
+    "fastapi",
     "opentelemetry-api",
     "opentelemetry-sdk",
     "opentelemetry-exporter-otlp",
     "opentelemetry-exporter-prometheus",
     "opentelemetry-instrumentation-pika",
-    "pika~=1.3.2",
+    "pika",
     "psutil",
     "python_multipart"
 ]

tools/create_exchange.py

@@ -57,7 +57,7 @@ def main():
     parser = argparse.ArgumentParser(description="Create a RabbitMQ exchange.")
     parser.add_argument(
         "--host",
-        default="localhost",
+        default="127.0.0.1",
         help="RabbitMQ host (default: localhost)",
     )