From a8d82b1e3d37655d2546ebdb19a4801e6ab3e41a Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 17 Apr 2025 13:20:56 +0100 Subject: [PATCH] #feat/1 - code cleanup for final release, part 1 --- amqp/adapter/amq_route_factory.py | 40 ++++---- amqp/adapter/cleverthis_service_adapter.py | 105 ++++++++++----------- amqp/adapter/logging_utils.py | 14 +++ amqp/rabbitmq/rabbit_mq_client.py | 3 +- amqp/service/delete_queues.py | 2 + 5 files changed, 89 insertions(+), 75 deletions(-) create mode 100644 amqp/adapter/logging_utils.py diff --git a/amqp/adapter/amq_route_factory.py b/amqp/adapter/amq_route_factory.py index db6b38c..a5c1da0 100644 --- a/amqp/adapter/amq_route_factory.py +++ b/amqp/adapter/amq_route_factory.py @@ -28,27 +28,27 @@ class AMQRouteFactory: :return: AMQRoute object """ try: - keys = data.split(RS) - if len(keys) >= 5: - i = 0 - component_name = keys[i] - i += 1 - exchange = keys[i] - i += 1 - queue = keys[i] if i < len(keys) else "" - i += 1 - routing_key = keys[i] if i < len(keys) else "" - i += 1 - timeout = int(keys[i]) if i < len(keys) else 0 - i += 1 - valid_until = int(keys[i]) if i < len(keys) else 0 + _keys = data.split(RS) + if len(_keys) >= 5: + _i = 0 + _component_name = _keys[_i] + _i += 1 + _exchange = _keys[_i] + _i += 1 + _queue = _keys[_i] if _i < len(_keys) else "" + _i += 1 + _routing_key = _keys[_i] if _i < len(_keys) else "" + _i += 1 + _timeout = int(_keys[_i]) if _i < len(_keys) else 0 + _i += 1 + _valid_until = int(_keys[_i]) if _i < len(_keys) else 0 return AMQRoute( - component_name=component_name, - exchange=exchange, - queue=queue, - key=routing_key, - timeout=timeout, - valid_until=valid_until + component_name=_component_name, + exchange=_exchange, + queue=_queue, + key=_routing_key, + timeout=_timeout, + valid_until=_valid_until ) logging.error( " [E] AMQRoute incorrect format. AMQRoute expects input as: " diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 526f5f3..be6c2ab 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -8,8 +8,7 @@ import json import logging import sys import traceback -import pathlib -from asyncio import Future, AbstractEventLoop +from asyncio import AbstractEventLoop from typing import Dict, List, Tuple, Callable, Coroutine, Any from urllib.parse import parse_qs, urlparse from xml.etree import ElementTree @@ -25,11 +24,9 @@ from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import DataMessage, DataResponse from amqp.adapter.multipart_parser import CleverMultiPartParser, process_form_data from amqp.adapter.file_uploader import publish_file_to_rabbitmq - -from IPython.core import debugger - from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE +from IPython.core import debugger debug = debugger.Pdb().set_trace @@ -54,10 +51,30 @@ class AMQMessage(DataMessage): self.connection = connection self.extra_data = data_message.extra_data if hasattr(data_message, 'extra_data') else {} + +def parse_get_url(url: str) -> Tuple[str, Dict[str, str]]: + """ + Parses an HTTP GET URL and returns (context_path, query_params_dict) + + Args: + url: The URL string to parse (e.g., "http://example.com/path?foo=1&bar=2&foo=3") + + Returns: + tuple: (context_path, query_params_dict) + - context_path: The path part of the URL (e.g., "/path") + - query_params_dict: Query parameters where values are str for single value or list of str for multivalue + """ + _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()} + return _parsed.path, _simplified_params + + 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. Args: message (DataMessage): Contains raw request and MIME headers. @@ -117,32 +134,36 @@ def parse_request_body(message: DataMessage): raise ValueError(f"Unsupported Content-Type: {content_type}") -async def convert_file_response( +async def _convert_file_response( obj: FileResponse, connection: AbstractRobustConnection, loop: AbstractEventLoop | None ) -> str: """ - Converts a FileResponse object to a JSON string containing the filename and stream. + 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 temporary dedicated queue. """ - async def wrap_connection_channel(connection: AbstractRobustConnection) -> tuple[AbstractRobustChannel, AbstractExchange, str]: - """ Wrap calls to RAbbitMQ API to make it Awaitable, so it can be run thread-safe with correct event loop """ + 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 correct event loop + # This is needed because the RabbitMQ API is not thread-safe and needs to be called in the context of + # 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) + _exchange_name: str = AMQ_REPLY_TO_EXCHANGE + _exchange = await _channel.get_exchange(name=_exchange_name, ensure=True) _queue = await _channel.declare_queue(exclusive=False) # Declare a unique, non-exclusive queue _queue_name = _queue.name await _queue.bind( - exchange=exchange_name, + exchange=_exchange_name, routing_key=_queue_name, # Use queue name as routing key ) - - logging.debug(f" [*] Publishing to exchange: {exchange_name}, q: {_queue_name}") + logging.debug(f" [*] Publishing to exchange: {_exchange_name}, q: {_queue_name}") return _channel, _exchange, _queue_name - # All calls to RabbittMQ API hav eto be called threadsafe, because those appear to be attached to parent + # All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached to parent # event loop, and need to execute at that event loop, not the current one - _future = asyncio.run_coroutine_threadsafe(wrap_connection_channel(connection), loop) + _future = asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop) while not _future.done(): await asyncio.sleep(0.1) # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute _channel, _exchange, _routing_key = _future.result() @@ -199,7 +220,7 @@ async def call_cleverthis_api( logging.info(f" [*] ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}") if isinstance(_result, FileResponse): - _json_result, _content_type = await convert_file_response( + _json_result, _content_type = await _convert_file_response( _result, message.connection, loop ), 'application/octet-stream' else: @@ -224,25 +245,6 @@ async def call_cleverthis_api( return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info) -def parse_get_url(url: str) -> Tuple[str, Dict[str, str]]: - """ - Parses an HTTP GET URL and returns (context_path, query_params_dict) - - Args: - url: The URL string to parse (e.g., "http://example.com/path?foo=1&bar=2&foo=3") - - Returns: - tuple: (context_path, query_params_dict) - - context_path: The path part of the URL (e.g., "/path") - - query_params_dict: Query parameters where values are str for single value or list of str for multivalue - """ - _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()} - return _parsed.path, _simplified_params - - # ============= CleverThisServiceAdapter ============= class CleverThisServiceAdapter: """ @@ -293,10 +295,10 @@ class CleverThisServiceAdapter: :return: AMQP response class with operation status code and optional error details. """ _auth_user: Any | None = None - amq_response: Any | None = None + _amq_response: Any | None = None - logging.warn( - f'Got here with: _auth_user={_auth_user} and require_authenticated_user={self.require_authenticated_user}') + logging.debug( + f' [*] on_message: _auth_user={_auth_user}, require_authenticated_user={self.require_authenticated_user}') if self.require_authenticated_user: _auth: str = message.headers.get('Authorization', '') logging.info(f"Auth: {_auth}") @@ -308,31 +310,26 @@ class CleverThisServiceAdapter: logging.warn(f'Token: {token}') _auth_user = await self.get_current_user(token) except Exception as error: - logging.error(f"Error getting user: {error}") - - # bearer_in_auth = 'Bearer' in _auth - # logging.warn(f'Got here2 with: Bearer_in_auth={bearer_in_auth}, _auth_user={_auth_user} and require_authenticated_user={self.require_authenticated_user}') - logging.warn( - f'Got here2 with: _auth_user={_auth_user} and require_authenticated_user={self.require_authenticated_user}') + logging.error(f" [E] on_message: Error getting user: {error}") if _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) + _amq_response = await self.get(message, _auth_user) elif method == 'put': - amq_response = await self.put(message, _auth_user) + _amq_response = await self.put(message, _auth_user) elif method == 'post': - amq_response = await self.post(message, _auth_user) + _amq_response = await self.post(message, _auth_user) elif method == 'head': - amq_response = await self.head(message, _auth_user) + _amq_response = await self.head(message, _auth_user) elif method == 'delete': - amq_response = await self.delete(message, _auth_user) + _amq_response = await self.delete(message, _auth_user) else: raise ValueError(f"Unexpected HTTP method: {message.method}") else: - amq_response = DataResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info) - logging.info(f" [x] ALL DONE in on_message id:{message.id}, resp code:{amq_response.response_code}") - return amq_response + _amq_response = DataResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info) + logging.info(f" [x] ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}") + return _amq_response # ================================================================================================== # ===================== C l e v e r T h i s A P I m e t h o d s ============================ diff --git a/amqp/adapter/logging_utils.py b/amqp/adapter/logging_utils.py new file mode 100644 index 0000000..838fce1 --- /dev/null +++ b/amqp/adapter/logging_utils.py @@ -0,0 +1,14 @@ +import logging +import sys +import traceback + + +def logging_error(msg: str, *args) -> None: + """ + Log an error message according to current logging for 'ERROR' level. + Add module name, line and function name where the error occurred, on single line. + Args: + msg (str): The error message to log. + """ + _tb = traceback.extract_tb(sys.exc_info()[2])[-1] + logging.error(f" [E] module '{_tb.filename}', line {_tb.lineno}, function '{_tb.name}': {msg}", *args) diff --git a/amqp/rabbitmq/rabbit_mq_client.py b/amqp/rabbitmq/rabbit_mq_client.py index 63f5996..0a635e1 100644 --- a/amqp/rabbitmq/rabbit_mq_client.py +++ b/amqp/rabbitmq/rabbit_mq_client.py @@ -5,6 +5,7 @@ from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange from pika.exchange_type import ExchangeType from amqp.adapter.data_message_factory import DataMessageFactory +from amqp.adapter.logging_utils import logging_error from amqp.model.model import DataMessage, AMQRoute from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE, DLQ_EXCHANGE, AMQ_RPC_EXCHANGE from amqp.router.router_consumer import RouterConsumer @@ -54,7 +55,7 @@ class RabbitMQClient: if not self.connection.is_closed: self.connection.close() except Exception as e: - logging.error(" [E] RabbitMQClient.PreDestroy cleanup - connection close error: %s", e) + logging_error(" [E] RabbitMQClient.PreDestroy cleanup - connection close error: %s", e) async def register_inbound_routes(self, route_mapping: str, callback): """ diff --git a/amqp/service/delete_queues.py b/amqp/service/delete_queues.py index a0a5547..a33b57d 100644 --- a/amqp/service/delete_queues.py +++ b/amqp/service/delete_queues.py @@ -38,3 +38,5 @@ def delete_queues_with_prefix_http(host='localhost', port=15672, username='sprin if __name__ == "__main__": delete_queues_with_prefix_http() + delete_queues_with_prefix_http(prefix='amq.gen--') + delete_queues_with_prefix_http(prefix='amq_')