From 5bb6597b5756a3371540ba0b1ca49a43fc835abf Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Wed, 7 May 2025 13:05:04 +0100 Subject: [PATCH] #40 - ensure the JSON format is handled correctly at all levels --- amqp/adapter/backpressure_handler.py | 18 ++++++------- amqp/adapter/cleverthis_service_adapter.py | 9 ++++++- amqp/adapter/data_parser.py | 7 ++++- amqp/adapter/data_response_factory.py | 10 ++++--- amqp/adapter/logging_utils.py | 31 ++++++++++++---------- amqp/adapter/serializer.py | 4 +++ amqp/model/model.py | 2 +- pyproject.toml | 10 +++---- tools/create_exchange.py | 2 +- 9 files changed, 58 insertions(+), 35 deletions(-) diff --git a/amqp/adapter/backpressure_handler.py b/amqp/adapter/backpressure_handler.py index 2d8a698..1fca9e2 100644 --- a/amqp/adapter/backpressure_handler.py +++ b/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 diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 126e794..37049ad 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/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" diff --git a/amqp/adapter/data_parser.py b/amqp/adapter/data_parser.py index d4daa94..7ba9425 100644 --- a/amqp/adapter/data_parser.py +++ b/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") ) diff --git a/amqp/adapter/data_response_factory.py b/amqp/adapter/data_response_factory.py index 3ec3f2d..5ee7908 100644 --- a/amqp/adapter/data_response_factory.py +++ b/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_cause=None, + error=error, + 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 diff --git a/amqp/adapter/logging_utils.py b/amqp/adapter/logging_utils.py index 1bea2e0..1046424 100644 --- a/amqp/adapter/logging_utils.py +++ b/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) diff --git a/amqp/adapter/serializer.py b/amqp/adapter/serializer.py index 5b1619c..86965f3 100644 --- a/amqp/adapter/serializer.py +++ b/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") diff --git a/amqp/model/model.py b/amqp/model/model.py index 1825195..0598461 100644 --- a/amqp/model/model.py +++ b/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, diff --git a/pyproject.toml b/pyproject.toml index 957c12b..01a71e0 100644 --- a/pyproject.toml +++ b/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", - "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" ] diff --git a/tools/create_exchange.py b/tools/create_exchange.py index e3da527..ae34507 100644 --- a/tools/create_exchange.py +++ b/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)", )