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

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(
+                # logging_info(
-                    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
+                #    "Backpressure: current CPU[pct]=%s, avg=%s, state=%s, last dataMsg=%s, last event=%s, last eventTime=%s",
-                    _current_cpu_usage,
+                #    _current_cpu_usage,
-                    _average_cpu_usage,
+                #    _average_cpu_usage,
-                    _cpu_usage_state,
+                #    _cpu_usage_state,
-                    self.last_data_message_time,
+                #    self.last_data_message_time,
-                    self.last_backpressure_event,
+                #    self.last_backpressure_event,
-                    self.last_backpressure_event_time,
+                #    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,8 +17,10 @@ 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
@@ -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"

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

@@ -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,8 +48,8 @@ class DataResponseFactory:
             response_code=response_code,
             content_type=content_type,
             response=body,
-            error=None,
+            error=error,
-            error_cause=None,
+            error_cause=error_cause,
             trace_info=trace_info,
         )
 
@@ -61,8 +63,10 @@ class DataResponseFactory:
             id,
             error_message.response_code,
             error_message.content_type,
-            error_message.serialze(),
+            error_message.serialize(),
             trace_info,
+            error=error_message.error,
+            error_cause=error_message.detail,
         )
 
     @staticmethod

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:
+    if not __verbose__:
-        logging.warning(f"{_thread_id} [W] {msg}", *args)
+        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:
+    if not __verbose__:
-        logging.info(f"{_thread_id} [*] {msg}", *args)
+        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

@@ -324,7 +324,7 @@ class AMQErrorMessage:
         self.detail = detail
         self.content_type = content_type
 
-    def serialze(self) -> bytes:
+    def serialize(self) -> bytes:
         return json.dumps(
             {
                 "response_code": self.response_code,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "amq_adapter"
-version = "0.2.28"
+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",
+    "aio-pika",
-    "aiohttp~=3.11.11",
+    "aiohttp",
-    "fastapi==0.114.2",
+    "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)",
     )