From 1de7afe4f58bbcd7c84db78bc19db60ca8c6f599 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Fri, 18 Apr 2025 23:49:10 +0100 Subject: [PATCH] feat/#1 - applied 'black' formatting --- amqp/adapter/amq_route_factory.py | 11 +- amqp/adapter/cleverthis_service_adapter.py | 224 +++++++++++++------ amqp/adapter/data_message_factory.py | 56 +++-- amqp/adapter/data_response_factory.py | 22 +- amqp/adapter/file_downloader.py | 88 +++++--- amqp/adapter/file_uploader.py | 40 ++-- amqp/adapter/logging_utils.py | 10 +- amqp/adapter/multipart_parser.py | 44 ++-- amqp/adapter/pydantic_serializer.py | 17 +- amqp/adapter/serializer.py | 26 +-- amqp/adapter/service_message_factory.py | 37 ++- amqp/adapter/test_amq_route_factory.py | 18 +- amqp/adapter/test_data_message_factory.py | 99 ++++++-- amqp/adapter/test_data_response_factory.py | 10 +- amqp/adapter/test_file_downloader.py | 9 +- amqp/adapter/test_pydantic_serializer.py | 10 +- amqp/adapter/test_service_message_factory.py | 22 +- amqp/adapter/trace_info_adapter.py | 5 +- amqp/config/amq_configuration.py | 75 +++++-- amqp/healthcheck.py | 6 +- amqp/model/model.py | 106 ++++++--- amqp/model/snowflake_id.py | 15 +- amqp/model/test_snowflake_id.py | 8 +- amqp/rabbitmq/rabbit_mq_client.py | 133 +++++++---- amqp/router/route_database.py | 10 +- amqp/router/router_base.py | 28 ++- amqp/router/router_consumer.py | 174 +++++++++----- amqp/router/router_producer.py | 209 ++++++++++------- amqp/router/test_route_database.py | 7 +- amqp/router/test_router_base.py | 9 +- amqp/router/utils.py | 8 +- amqp/service/amq_message_handler.py | 207 +++++++++++------ amqp/service/amq_service.py | 31 +-- amqp/service/delete_queues.py | 14 +- amqp/service/tracing.py | 14 +- 35 files changed, 1239 insertions(+), 563 deletions(-) diff --git a/amqp/adapter/amq_route_factory.py b/amqp/adapter/amq_route_factory.py index 694b49c..9f221a0 100644 --- a/amqp/adapter/amq_route_factory.py +++ b/amqp/adapter/amq_route_factory.py @@ -12,7 +12,7 @@ NULL_ROUTE = { "queue": None, "routingKey": "#", "timeout": 1, - "validUntil": 0 + "validUntil": 0, } @@ -49,18 +49,21 @@ class AMQRouteFactory: queue=_queue, key=_routing_key, timeout=_timeout, - valid_until=_valid_until + valid_until=_valid_until, ) logging_error( "AMQRoute incorrect format. AMQRoute expects input as: " "'Component-Name:AMQExchange-Name:Queue-Name:Routing-Key:Timeout:Valid-Until', " - "got: ['%s']", data + "got: ['%s']", + data, ) except Exception as err: logging_error( "%s: AMQRoute incorrect format. AMQRoute expects input as: " "'Component-Name:AMQExchange-Name:Queue-Name:Routing-Key:Timeout:Valid-Until', " - "got: ['%s']", err, data + "got: ['%s']", + err, + data, ) return NULL_ROUTE diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 65ab767..22c5774 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -2,6 +2,7 @@ The layer that interfaces with the CleverThis service that this Adapter is integrating with. Methods here will invoke the actual CleverThis Service code, which are passed as Callable object. """ + import asyncio import inspect import json @@ -13,7 +14,12 @@ from typing import Dict, List, Tuple, Callable, Coroutine, Any from urllib.parse import parse_qs, urlparse from xml.etree import ElementTree -from aio_pika.abc import AbstractRobustChannel, AbstractRobustConnection, AbstractRobustExchange, AbstractExchange +from aio_pika.abc import ( + AbstractRobustChannel, + AbstractRobustConnection, + AbstractRobustExchange, + AbstractExchange, +) from aiohttp import ClientSession, ClientResponse from dataclasses import dataclass from fastapi import HTTPException @@ -28,6 +34,7 @@ from amqp.adapter.file_uploader import publish_file_to_rabbitmq from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE from IPython.core import debugger + debug = debugger.Pdb().set_trace @@ -50,7 +57,9 @@ class AMQMessage(DataMessage): self.base64_body = data_message.base64_body # Add the new field self.connection = connection - self.extra_data = data_message.extra_data if hasattr(data_message, 'extra_data') else {} + 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]]: @@ -68,7 +77,9 @@ def parse_get_url(url: str) -> Tuple[str, Dict[str, str]]: _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()} + _simplified_params = { + k: v[0] if len(v) == 1 else v for k, v in _query_params.items() + } return _parsed.path, _simplified_params @@ -82,7 +93,7 @@ def parse_request_body(message: DataMessage): Returns: dict: Parsed data as a dictionary. """ - _content_type = message.headers.get('Content-Type', '') + _content_type = message.headers.get("Content-Type", "") if not _content_type: raise ValueError("Content-Type header is required") content_type = _content_type[0] @@ -97,10 +108,13 @@ def parse_request_body(message: DataMessage): # Parse URL-encoded form data elif content_type == "application/x-www-form-urlencoded": try: - _form_data: dict = {k: v[0] if len(v) == 1 else v for k, v in parse_qs(message.body()).items()} + _form_data: dict = { + k: v[0] if len(v) == 1 else v + for k, v in parse_qs(message.body()).items() + } if len(_form_data) == 0 and len(message.body()) > 0: _form_data = json.loads(message.body()) - if _form_data['files'] and _form_data['form']: + if _form_data["files"] and _form_data["form"]: _form_data = process_form_data(_form_data) return _form_data except Exception as e: @@ -135,14 +149,15 @@ def parse_request_body(message: DataMessage): async def _convert_file_response( obj: FileResponse, connection: AbstractRobustConnection, - loop: AbstractEventLoop | None + loop: AbstractEventLoop | None, ) -> str: """ 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_amq_api_calls( - connection: AbstractRobustConnection + 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 @@ -150,39 +165,53 @@ async def _convert_file_response( _channel = await connection.channel() _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 = await _channel.declare_queue( + exclusive=False + ) # Declare a unique, non-exclusive queue _queue_name = _queue.name await _queue.bind( 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 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_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 + 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() # publish_file also needs RabbitMQ API, ensure the loop is passed as argument - (_stream_name, _file_size) = await publish_file_to_rabbitmq(_exchange, obj.path, _routing_key, loop) + (_stream_name, _file_size) = await publish_file_to_rabbitmq( + _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) - return json.dumps({'files': [ - {'filename': obj.filename, - 'mediaType': 'application/octet-stream', - 'streamName': _stream_name, - 'fileSize': _file_size - }] - }) + return json.dumps( + { + "files": [ + { + "filename": obj.filename, + "mediaType": "application/octet-stream", + "streamName": _stream_name, + "fileSize": _file_size, + } + ] + } + ) def is_likely_json(text: str) -> bool: - """ Quick test to see if should convert the response string to JSON format. """ + """Quick test to see if should convert the response string to JSON format.""" _text = text.strip() - return (_text.startswith('{') and _text.endswith('}')) or \ - (_text.startswith('[') and _text.endswith(']')) + return (_text.startswith("{") and _text.endswith("}")) or ( + _text.startswith("[") and _text.endswith("]") + ) async def call_cleverthis_api( @@ -190,7 +219,7 @@ async def call_cleverthis_api( args: Any, api_method: Callable[[Any, Any], Coroutine], success_code: int, - loop: AbstractEventLoop | None = None + loop: AbstractEventLoop | None = None, ) -> DataResponse: """ Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary. @@ -220,41 +249,69 @@ async def call_cleverthis_api( else: _verified_args[arg.name] = args[arg.name] else: - logging.info(f"{api_method.__name__}: Missing required argument: {arg.name}") + logging.info( + f"{api_method.__name__}: Missing required argument: {arg.name}" + ) logging.info( - f" [*] INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}") + f" [*] INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}" + ) _result: Any = await api_method(**_verified_args) - logging.info(f" [*] ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}") + logging.info( + f" [*] ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}" + ) if isinstance(_result, FileResponse): - _json_result, _content_type = await _convert_file_response( - _result, message.connection, loop - ), 'application/octet-stream' + _json_result, _content_type = ( + await _convert_file_response(_result, message.connection, loop), + "application/octet-stream", + ) else: _json_result = _result - _content_type = 'application/json' + _content_type = "application/json" - _binary_result = _json_result.encode('utf-8') if hasattr( - _json_result, 'encode' - ) else _json_result.body if hasattr( - _json_result, 'body' - ) else _json_result.read() if hasattr(_json_result, 'read') else _json_result - logging.info(f" [x] ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}") - return DataResponseFactory.of(message.id, success_code, _content_type, _binary_result, message.trace_info) + _binary_result = ( + _json_result.encode("utf-8") + if hasattr(_json_result, "encode") + else ( + _json_result.body + if hasattr(_json_result, "body") + else ( + _json_result.read() + if hasattr(_json_result, "read") + else _json_result + ) + ) + ) + logging.info( + f" [x] ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}" + ) + return DataResponseFactory.of( + message.id, success_code, _content_type, _binary_result, message.trace_info + ) except HTTPException as 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(message.id, http_error.status_code, "application/json", - _content.encode('utf-8'), message.trace_info) + return DataResponseFactory.of( + message.id, + http_error.status_code, + "application/json", + _content.encode("utf-8"), + message.trace_info, + ) except Exception as error: logging_error(f"Service endpoint invocation failed: {error}") _content = str(error) if not is_likely_json(_content): _content = '{"error": "' + _content + '"}' - return DataResponseFactory.of(message.id, 500, "application/json", - _content.encode('utf-8'), message.trace_info) + return DataResponseFactory.of( + message.id, + 500, + "application/json", + _content.encode("utf-8"), + message.trace_info, + ) # ============= CleverThisServiceAdapter ============= @@ -266,7 +323,9 @@ class CleverThisServiceAdapter: and override the methods to handle the specific API calls. """ - def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None): + def __init__( + self, amq_configuration: AMQConfiguration, session: ClientSession = None + ): """ Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP session. The HTTP session is for the case when the adapter uses loose coupling with the CleverThis service via @@ -277,8 +336,13 @@ class CleverThisServiceAdapter: self.service_host = self.amq_configuration.dispatch.service_host self.service_port = self.amq_configuration.dispatch.service_port self.loop: AbstractEventLoop | None = None - self.require_authenticated_user = eval( - self.amq_configuration.amq_adapter.require_authenticated_user.capitalize()) if self.amq_configuration.amq_adapter.require_authenticated_user.strip() else True + self.require_authenticated_user = ( + eval( + self.amq_configuration.amq_adapter.require_authenticated_user.capitalize() + ) + if self.amq_configuration.amq_adapter.require_authenticated_user.strip() + else True + ) def get_content_type(self, message_headers: Dict[str, List[str]]) -> str: """ @@ -287,9 +351,9 @@ class CleverThisServiceAdapter: return: The content type of the message. """ for header, values in message_headers.items(): - if header.lower() == 'content-type': - return values[0].split(';')[0] - return 'application/json' + if header.lower() == "content-type": + return values[0].split(";")[0] + return "application/json" async def get_current_user(self, token: str) -> object: """ @@ -310,37 +374,46 @@ class CleverThisServiceAdapter: _amq_response: Any | None = None logging.debug( - f' [*] on_message: _auth_user={_auth_user}, require_authenticated_user={self.require_authenticated_user}') + 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', '') + _auth: str = message.headers.get("Authorization", "") logging.info(f"Auth: {_auth}") if isinstance(_auth, List): _auth = _auth[0] - if 'Bearer' in _auth: + if "Bearer" in _auth: try: - token = _auth.split(' ')[1] - logging.warn(f'Token: {token}') + token = _auth.split(" ")[1] + logging.warn(f"Token: {token}") _auth_user = await self.get_current_user(token) except Exception as error: logging_error(f"on_message: Error getting user: {error}") - if _auth_user or not self.require_authenticated_user or message.path.endswith('/login'): + if ( + _auth_user + or not self.require_authenticated_user + or message.path.endswith("/login") + ): method = message.method.lower() - if method == 'get': + if method == "get": _amq_response = await self.get(message, _auth_user) - elif method == 'put': + elif method == "put": _amq_response = await self.put(message, _auth_user) - elif method == 'post': + elif method == "post": _amq_response = await self.post(message, _auth_user) - elif method == 'head': + elif method == "head": _amq_response = await self.head(message, _auth_user) - elif method == 'delete': + elif method == "delete": _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}") + _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 # ================================================================================================== @@ -359,14 +432,25 @@ class CleverThisServiceAdapter: # unmatched path - try to invoke the service directly on this path url = f"http://{self.service_host}:{self.service_port}{message.path}" headers = {**message.headers, **message.trace_info} - logging.info(f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}") + logging.info( + f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}" + ) if self.session: _http_resp: ClientResponse = await self.session.get(url, headers=headers) - return DataResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type, - await _http_resp.read(), - message.trace_info) - return DataResponseFactory.of(message.id, 404, "text/plain", - str("Destination location does not exists").encode('utf-8'), message.trace_info) + return DataResponseFactory.of( + message.id, + _http_resp.status, + _http_resp.content_type, + await _http_resp.read(), + message.trace_info, + ) + return DataResponseFactory.of( + message.id, + 404, + "text/plain", + str("Destination location does not exists").encode("utf-8"), + message.trace_info, + ) async def put(self, message: DataMessage, auth_user: Any) -> DataResponse: """ @@ -383,7 +467,7 @@ class CleverThisServiceAdapter: data=message.body, headers={**message.headers, **message.trace_info}, ), - auth_user + auth_user, ) async def post(self, message: DataMessage, auth_user: Any) -> DataResponse: @@ -401,10 +485,12 @@ class CleverThisServiceAdapter: data=message.body, headers={**message.headers, **message.trace_info}, ), - auth_user + auth_user, ) - async def handle_possible_form(self, message: DataMessage, request_coroutine, auth_user: Any) -> DataResponse: + async def handle_possible_form( + self, message: DataMessage, request_coroutine, auth_user: Any + ) -> DataResponse: _args = parse_request_body(message) return await request_coroutine diff --git a/amqp/adapter/data_message_factory.py b/amqp/adapter/data_message_factory.py index 0e03f6c..722e0bc 100644 --- a/amqp/adapter/data_message_factory.py +++ b/amqp/adapter/data_message_factory.py @@ -11,7 +11,12 @@ from typing import Dict, List from amqp.model.model import DataMessage from amqp.model.snowflake_id import SnowflakeId -from amqp.adapter.serializer import map_with_list_as_string, map_as_string, parse_map_list_string, parse_map_string +from amqp.adapter.serializer import ( + map_with_list_as_string, + map_as_string, + parse_map_list_string, + parse_map_string, +) from amqp.router.utils import sanitize_as_rabbitmq_name from amqp.adapter.file_downloader import download_buffer @@ -65,19 +70,24 @@ class DataMessageFactory: self.snowflake_sequence = 0 self.snowflake_last_timestamp = timestamp - time = (timestamp - 1000 * int(DataMessageFactory.snowflake_epoch)) + time = timestamp - 1000 * int(DataMessageFactory.snowflake_epoch) return SnowflakeId( time >> (64 - self.SEQUENCE_BITS - self.MACHINE_ID_BITS), - ((time << (self.SEQUENCE_BITS + self.MACHINE_ID_BITS)) | self.machine_id | self.snowflake_sequence) & 0xFFFFFFFFFFFFFFFF + ( + (time << (self.SEQUENCE_BITS + self.MACHINE_ID_BITS)) + | self.machine_id + | self.snowflake_sequence + ) + & 0xFFFFFFFFFFFFFFFF, ) def create_request_message( - self, - method: str, - domain: str, - path: str, - headers: Dict[str, List[str]], - message, + self, + method: str, + domain: str, + path: str, + headers: Dict[str, List[str]], + message, ) -> DataMessage: """ Encapsulate / hide the system level inputs, create the AMQ message from business relevant input @@ -89,7 +99,9 @@ class DataMessageFactory: :return: DataMessage object """ serializable_headers = headers.copy() if headers else {} - payload = base64.b64encode(message.encode('utf-8') if isinstance(message, str) else message) + payload = base64.b64encode( + message.encode("utf-8") if isinstance(message, str) else message + ) trace_info = {} return DataMessage( self.MAGIC_BYTE_HTTP_REQUEST, @@ -119,8 +131,12 @@ class DataMessageFactory: :return: the message timestamp """ bits = _id.get_bits() - low = bits[0] >> (DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS) - high = bits[1] << (64 - DataMessageFactory.SEQUENCE_BITS - DataMessageFactory.MACHINE_ID_BITS) + low = bits[0] >> ( + DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS + ) + high = bits[1] << ( + 64 - DataMessageFactory.SEQUENCE_BITS - DataMessageFactory.MACHINE_ID_BITS + ) return (high | low) + DataMessageFactory.snowflake_epoch * 1000 @staticmethod @@ -152,13 +168,13 @@ class DataMessageFactory: id_bytes = str(message.id).encode() prolog = ( f"|m={message.method}|d={message.domain}|p={message.path}|h={{{map_with_list_as_string(message.headers)}}}|t={{{map_as_string(message.trace_info)}}}|b=" - ).encode('utf-8') + ).encode("utf-8") header_length = 1 + len(id_bytes) + len(prolog) - prolog_len = header_length.to_bytes(2, byteorder='big') + prolog_len = header_length.to_bytes(2, byteorder="big") msg_length = 2 + header_length + len(message.base64_body) with BytesIO(bytearray(msg_length)) as baos: baos.write(prolog_len) - baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8')) + baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8")) baos.write(id_bytes) baos.write(prolog) baos.write(message.base64_body) @@ -172,7 +188,7 @@ class DataMessageFactory: :return: DataMessage instance """ prolog_len = (input_bytes[0] << 8) + input_bytes[1] - metadata = input_bytes[2:prolog_len + 2].decode() + metadata = input_bytes[2 : prolog_len + 2].decode() pattern = re.compile( f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}" @@ -197,7 +213,7 @@ class DataMessageFactory: path = match.group(4) headers_str = match.group(5) trace_info_str = match.group(6) - body_b64 = input_bytes[prolog_len + 2:] + body_b64 = input_bytes[prolog_len + 2 :] headers = parse_map_list_string(headers_str) trace_info = parse_map_string(trace_info_str) @@ -223,9 +239,11 @@ class DataMessageFactory: """ # async implementation message_data = json.loads(stream.decode()) - req_info = message_data.get('req-info', '') + req_info = message_data.get("req-info", "") if req_info: - (body, eof) = await download_buffer(loop=loop, rabbitmq_url=rabbitmq_url, queue_name=req_info) + (body, eof) = await download_buffer( + loop=loop, rabbitmq_url=rabbitmq_url, queue_name=req_info + ) if eof == 1: return DataMessageFactory.from_bytes(body) return None diff --git a/amqp/adapter/data_response_factory.py b/amqp/adapter/data_response_factory.py index 1c29aaa..5a7048d 100644 --- a/amqp/adapter/data_response_factory.py +++ b/amqp/adapter/data_response_factory.py @@ -25,7 +25,13 @@ class DataResponseFactory: ) @staticmethod - def of(id: SnowflakeId, response_code: int, content_type: str, body: bytes, trace_info: dict[str,str]) -> DataResponse: + def of( + id: SnowflakeId, + response_code: int, + content_type: str, + body: bytes, + trace_info: dict[str, str], + ) -> DataResponse: """ Create the DataResponse message supplying all key values :param id: SnowflakeID @@ -42,7 +48,7 @@ class DataResponseFactory: response=body, error=None, error_cause=None, - trace_info=trace_info + trace_info=trace_info, ) @staticmethod @@ -52,16 +58,16 @@ class DataResponseFactory: :param response: Object to serialize :return: byte array as serialized message """ - id_bytes = response.id.encode('utf-8') + id_bytes = response.id.encode("utf-8") prolog = ( f"|r={response.response_code}|c={response.content_type}|e={response.error}|f={response.error_cause}|t={{{map_as_string(response.trace_info)}}}|b=" - ).encode('utf-8') + ).encode("utf-8") header_length = 1 + len(id_bytes) + len(prolog) - prolog_len = header_length.to_bytes(2, byteorder='big') + prolog_len = header_length.to_bytes(2, byteorder="big") msg_length = 2 + header_length + len(response.response) with BytesIO(bytearray(msg_length)) as baos: baos.write(prolog_len) - baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode('utf-8')) + baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8")) baos.write(id_bytes) baos.write(prolog) baos.write(base64.b64encode(response.response)) @@ -75,7 +81,7 @@ class DataResponseFactory: :return: DataResponse object """ prolog_len = (input_bytes[0] << 8) + input_bytes[1] - metadata = input_bytes[2:prolog_len + 2].decode() + metadata = input_bytes[2 : prolog_len + 2].decode() pattern = re.compile( f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}" @@ -100,7 +106,7 @@ class DataResponseFactory: error = match.group(4) error_cause = match.group(5) trace_info_str = match.group(6) - body_b64 = input_bytes[prolog_len + 2:] + body_b64 = input_bytes[prolog_len + 2 :] trace_info = parse_map_string(trace_info_str) diff --git a/amqp/adapter/file_downloader.py b/amqp/adapter/file_downloader.py index bea02dd..70963b3 100644 --- a/amqp/adapter/file_downloader.py +++ b/amqp/adapter/file_downloader.py @@ -52,7 +52,9 @@ def ensure_directory_exists(filepath) -> str: version += 1 -async def download_file(loop, rabbitmq_url, file_info, message_data, output_dir) -> (int, int): +async def download_file( + loop, rabbitmq_url, file_info, message_data, output_dir +) -> (int, int): """ Asynchronously downloads a file from RabbitMQ based on the provided file information. takes the file_info and message_data and output_dir, creates queue consumer and waits for the file to be downloaded @@ -64,17 +66,23 @@ async def download_file(loop, rabbitmq_url, file_info, message_data, output_dir) 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'] - queue_name = file_info['queue_name'] + filename = file_info["filename"] + size = file_info["size"] + stream_name = file_info["streamName"] + queue_name = file_info["queue_name"] filepath = os.path.join(output_dir, filename) - filepath = ensure_directory_exists(filepath) # if filename exists, create a new (next) version - download_locations[file_info['key']] = filepath - filename = os.path.basename(filepath) # if a new version was created, ensure to use that new version - file_info['filename'] = filename + filepath = ensure_directory_exists( + filepath + ) # if filename exists, create a new (next) version + download_locations[file_info["key"]] = filepath + filename = os.path.basename( + filepath + ) # if a new version was created, ensure to use that new version + file_info["filename"] = filename - logging.info(f"Downloading file: {filename} (size: {size} bytes) from stream: {stream_name}:{queue_name} to {filepath}") + logging.info( + f"Downloading file: {filename} (size: {size} bytes) from stream: {stream_name}:{queue_name} to {filepath}" + ) async def wrap_rabbit_mq_calls() -> tuple[int, int] | None: connection = await connect_robust(rabbitmq_url, loop=loop) @@ -87,10 +95,10 @@ async def download_file(loop, rabbitmq_url, file_info, message_data, output_dir) async with queue.iterator() as queue_iter: async for message in queue_iter: async with message.process(): - with open(filepath, 'ab') as f: + with open(filepath, "ab") as f: f.write(message.body) _file_size = _file_size + message.body_size - _eof = int(message.headers.get('eof', END_OF_FILE)) + _eof = int(message.headers.get("eof", END_OF_FILE)) if _eof != IN_PROGRESS: break except Exception as e: @@ -101,7 +109,9 @@ async def download_file(loop, rabbitmq_url, file_info, message_data, output_dir) return _file_size, _eof if queue_name: - _future: Future = asyncio.run_coroutine_threadsafe(wrap_rabbit_mq_calls(), loop=loop) + _future: Future = asyncio.run_coroutine_threadsafe( + wrap_rabbit_mq_calls(), loop=loop + ) while not _future.done(): await asyncio.sleep(0.05) _local_file_size, _local_eof = _future.result() @@ -109,6 +119,7 @@ async def download_file(loop, rabbitmq_url, file_info, message_data, output_dir) return 0, CANCELLED + async def download_buffer(loop, rabbitmq_url, queue_name) -> (bytearray, int): """ Asynchronously downloads a bytearay from RabbitMQ based on the provided file information. @@ -120,6 +131,7 @@ async def download_buffer(loop, rabbitmq_url, queue_name) -> (bytearray, int): """ logging.info(f"Downloading buffer from stream: {queue_name}") if queue_name: + async def wrap_rabbit_mq_calls() -> tuple[bytearray, int]: _connection = await connect_robust(rabbitmq_url, loop=loop) _eof = IN_PROGRESS @@ -127,12 +139,14 @@ async def download_buffer(loop, rabbitmq_url, queue_name) -> (bytearray, int): try: channel: AbstractRobustChannel = await _connection.channel() queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False) - logging.info(f"Awaiting all buffer chunks being downloaded for {queue_name}") + logging.info( + f"Awaiting all buffer chunks being downloaded for {queue_name}" + ) async with queue.iterator() as queue_iter: async for message in queue_iter: async with message.process(): _res.extend(message.body) - _eof = message.headers.get('eof', END_OF_FILE) + _eof = message.headers.get("eof", END_OF_FILE) if _eof != IN_PROGRESS: break logging.info(f"Finished downloading buffer {queue_name}") @@ -142,7 +156,9 @@ async def download_buffer(loop, rabbitmq_url, queue_name) -> (bytearray, int): await _connection.close() return _res, _eof - _future: Future = asyncio.run_coroutine_threadsafe(wrap_rabbit_mq_calls(), loop=loop) + _future: Future = asyncio.run_coroutine_threadsafe( + wrap_rabbit_mq_calls(), loop=loop + ) while not _future.done(): await asyncio.sleep(0.05) _local_res, _local_eof = _future.result() @@ -155,7 +171,7 @@ async def on_message( json_data: bytes, rabbitmq_url: str, loop, - output_dir="downloaded_files" + output_dir="downloaded_files", ) -> defaultdict: """ Asynchronous callback function for handling messages from the 'amq-cleverswarm' queue. @@ -170,35 +186,47 @@ async def on_message( _amq_message_data = json.loads(json_data.decode()) files_to_download = [] - for _key, _file_type in _amq_message_data['files'].items(): + for _key, _file_type in _amq_message_data["files"].items(): if len(_file_type) > 0: files_info = _file_type[0] - stream_name = files_info['streamName'] - files_info['queue_name'] = _message_data.get(stream_name, "") if stream_name else "" - files_info['key'] = _key + stream_name = files_info["streamName"] + files_info["queue_name"] = ( + _message_data.get(stream_name, "") if stream_name else "" + ) + files_info["key"] = _key files_to_download.append(files_info) if not files_to_download: - logging.info(" [%s] No files to download in this message.", message.delivery_tag) + logging.info( + " [%s] No files to download in this message.", message.delivery_tag + ) # await message.ack() return False # Create a task for each file download _tasks = [] for file_info in files_to_download: - logging.info(f" [{message.delivery_tag}] about to create task for {file_info}") - _task = asyncio.create_task(download_file(loop, rabbitmq_url, file_info, _message_data, output_dir)) + logging.info( + f" [{message.delivery_tag}] about to create task for {file_info}" + ) + _task = asyncio.create_task( + download_file(loop, rabbitmq_url, file_info, _message_data, output_dir) + ) _tasks.append(_task) - logging.info(f" [{message.delivery_tag}] Waiting for all files to download...gather(*tasks)") - await asyncio.gather(*_tasks) # Wait for all downloads to complete + logging.info( + f" [{message.delivery_tag}] Waiting for all files to download...gather(*tasks)" + ) + await asyncio.gather(*_tasks) # Wait for all downloads to complete # After all files are downloaded, acknowledge the original message logging.info(f"All files downloaded. d-tag: {message.delivery_tag}") - #logging.info(f"All files downloaded. Acknowledging the main message d-tag: {message.delivery_tag}") - #await message.ack() - #logging.info(f"Message d-tag: {message.delivery_tag} acknowledged.") + # logging.info(f"All files downloaded. Acknowledging the main message d-tag: {message.delivery_tag}") + # await message.ack() + # logging.info(f"Message d-tag: {message.delivery_tag} acknowledged.") except Exception as e: logging_error(f"Building DataMessage: {e}") - asyncio.run_coroutine_threadsafe(message.ack(), loop=loop) # ACK -> ignore the message (do not redeliver) + asyncio.run_coroutine_threadsafe( + message.ack(), loop=loop + ) # ACK -> ignore the message (do not redeliver) return download_locations diff --git a/amqp/adapter/file_uploader.py b/amqp/adapter/file_uploader.py index ea24b13..ca84a68 100644 --- a/amqp/adapter/file_uploader.py +++ b/amqp/adapter/file_uploader.py @@ -20,7 +20,7 @@ async def publish_file_to_rabbitmq( routing_key: str, loop: AbstractEventLoop, max_chunk_size: int = 2**20, # 1MB default chunk size - content_type: str = 'application/octet-stream', # Default content type + content_type: str = "application/octet-stream", # Default content type delivery_mode: DeliveryMode = DeliveryMode.PERSISTENT, ) -> tuple[str, int]: """ @@ -45,7 +45,7 @@ async def publish_file_to_rabbitmq( raise FileNotFoundError(f"File not found: {file_path}") _file_size = os.path.getsize(file_path) - with open(file_path, 'rb') as f: + with open(file_path, "rb") as f: _offset = 0 while True: _chunk = f.read(max_chunk_size) @@ -57,22 +57,36 @@ async def publish_file_to_rabbitmq( content_type=content_type, delivery_mode=delivery_mode, headers={ - 'file_name': file_path.name, - 'pos': _offset, - 'total_size': _file_size, - 'chunk_size': len(_chunk), - 'eof': END_OF_FILE if max_chunk_size + _offset >= _file_size else IN_PROGRESS - } + "file_name": file_path.name, + "pos": _offset, + "total_size": _file_size, + "chunk_size": len(_chunk), + "eof": ( + END_OF_FILE + if max_chunk_size + _offset >= _file_size + else IN_PROGRESS + ), + }, + ) + logging.debug( + f"Going to Publish chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}" + ) + _future = asyncio.run_coroutine_threadsafe( + exchange.publish(message, routing_key=routing_key), loop ) - logging.debug(f"Going to Publish chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}") - _future = asyncio.run_coroutine_threadsafe(exchange.publish(message, routing_key=routing_key), loop) while not _future.done(): # sleep allows coroutine to execute. must wait for completion to ensure all chunks written in right order await asyncio.sleep(0.02) - logging.debug(f"Waiting for chunk {_offset}/{_file_size} future= {_future}") + logging.debug( + f"Waiting for chunk {_offset}/{_file_size} future= {_future}" + ) _offset += len(_chunk) - logging.debug(f"Published chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}") + logging.debug( + f"Published chunk {_offset}/{_file_size} to {exchange.name}:{routing_key}" + ) - logging_debug(f"File {file_path} published to RabbitMQ exchange {exchange.name}, queue {routing_key}") + logging_debug( + f"File {file_path} published to RabbitMQ exchange {exchange.name}, queue {routing_key}" + ) return routing_key, _file_size diff --git a/amqp/adapter/logging_utils.py b/amqp/adapter/logging_utils.py index 879f0a8..89cd5e2 100644 --- a/amqp/adapter/logging_utils.py +++ b/amqp/adapter/logging_utils.py @@ -13,7 +13,10 @@ def logging_error(msg: str, *args) -> None: _exec_info = sys.exc_info()[2] if _exec_info: _tb = traceback.extract_tb(sys.exc_info()[2])[-1] - logging.error(f" [E] '{msg}' ---> '{_tb.filename}': {_tb.lineno}, func: '{_tb.name}'", *args) + logging.error( + f" [E] '{msg}' ---> '{_tb.filename}': {_tb.lineno}, func: '{_tb.name}'", + *args, + ) else: logging.error(f" [E] {msg}", *args) @@ -28,6 +31,9 @@ def logging_debug(msg: str, *args) -> None: _exec_info = sys.exc_info()[2] if _exec_info: _tb = traceback.extract_tb(sys.exc_info()[2])[-1] - logging.debug(f" [DBG] '{msg}' ---> '{_tb.filename}': {_tb.lineno}, func: '{_tb.name}'", *args) + logging.debug( + f" [DBG] '{msg}' ---> '{_tb.filename}': {_tb.lineno}, func: '{_tb.name}'", + *args, + ) else: logging.debug(f" [DBG] {msg}", *args) diff --git a/amqp/adapter/multipart_parser.py b/amqp/adapter/multipart_parser.py index 2b0d2f3..dba985e 100644 --- a/amqp/adapter/multipart_parser.py +++ b/amqp/adapter/multipart_parser.py @@ -33,12 +33,12 @@ def process_form_data(json_input) -> dict: result = {} # Process form data - take first element of each list - for key, value in data['form'].items(): + for key, value in data["form"].items(): if value: # Only process if list is not empty result[key] = value[0] # Override with files data where available - for key, value in data['files'].items(): + for key, value in data["files"].items(): if value: # Only override if files data is non-empty result[key] = value[0] @@ -52,32 +52,45 @@ class CleverMultiPartParser: self.is_json = False self.is_valid = True # Convert each list of strings to a single string joined by newline, then to bytes - headers = {key: "\n".join(value).encode("utf-8") for key, value in message.headers.items()} + headers = { + key: "\n".join(value).encode("utf-8") + for key, value in message.headers.items() + } content_type: str | bytes | None = headers.get("Content-Type") content_type, params = multipart.parse_options_header(content_type) - if content_type.decode('utf-8') in ["application/octet-stream", - "multipart/form-data", - "application/x-www-form-urlencoded", - "application/x-url-encoded"]: + if content_type.decode("utf-8") in [ + "application/octet-stream", + "multipart/form-data", + "application/x-www-form-urlencoded", + "application/x-url-encoded", + ]: try: - if hasattr(message, 'extra_data') and isinstance(message.extra_data, dict) and len(message.extra_data) > 0: + if ( + hasattr(message, "extra_data") + and isinstance(message.extra_data, dict) + and len(message.extra_data) > 0 + ): try: self.form_data = json.loads(message.body()) self.is_json = True - if self.form_data['files'] and self.form_data['form']: + if self.form_data["files"] and self.form_data["form"]: self.form_data = process_form_data(self.form_data) except Exception as jsonerror: logging_error(f"Parsing JSON: {jsonerror}") self.is_valid = False else: - python_multipart.parse_form(headers, io.BytesIO(message.body()), self.on_field, self.on_file) + python_multipart.parse_form( + headers, io.BytesIO(message.body()), self.on_field, self.on_file + ) self.is_multipart = True except Exception as e: - logging_error(f"Error parsing multipart/form-data: {e}. Trying parsing as JSON.") + logging_error( + f"Error parsing multipart/form-data: {e}. Trying parsing as JSON." + ) try: self.form_data = json.loads(message.body()) self.is_json = True - if self.form_data['files'] and self.form_data['form']: + if self.form_data["files"] and self.form_data["form"]: self.form_data = process_form_data(self.form_data) except Exception as jsonerror: logging_error(f"Parsing JSON: {jsonerror}") @@ -87,11 +100,12 @@ class CleverMultiPartParser: def on_field(self, field: Field): logging_debug(field) - self.form_data[field.field_name.decode('utf-8')] = field.value + self.form_data[field.field_name.decode("utf-8")] = field.value logging_debug(self.form_data) def on_file(self, file: File): logging_debug(file) - self.form_data[file.field_name.decode('utf-8')] = \ - UploadFile(file.file_object, size=file.size, filename=file.file_name) + self.form_data[file.field_name.decode("utf-8")] = UploadFile( + file.file_object, size=file.size, filename=file.file_name + ) logging_debug(self.form_data) diff --git a/amqp/adapter/pydantic_serializer.py b/amqp/adapter/pydantic_serializer.py index 1233b84..6bee309 100644 --- a/amqp/adapter/pydantic_serializer.py +++ b/amqp/adapter/pydantic_serializer.py @@ -63,7 +63,7 @@ def deserialize_object(data: str) -> BaseModel: An R2RSerializable object. """ class_name, dict_str = data.split(":", 1) - dict_str = dict_str.strip('{}') + dict_str = dict_str.strip("{}") pairs = dict_str.split(",") if dict_str else [] obj_dict: Dict[str, Any] = {} @@ -75,8 +75,8 @@ def deserialize_object(data: str) -> BaseModel: obj_dict[k] = _convert_value(v) logging_debug(f"Deserialized object: {class_name}") - if '[' in class_name: - class_name = class_name.split('[')[0] + if "[" in class_name: + class_name = class_name.split("[")[0] logging_debug(f"trying object: {class_name}") cls = globals().get(class_name) if cls is None: @@ -89,7 +89,7 @@ def _convert_value(v: str) -> Any: """ Helper function to convert a string to its appropriate Python type. """ - if v == 'None': + if v == "None": return None elif _is_uuid_string(v): return uuid.UUID(v) @@ -97,8 +97,13 @@ def _convert_value(v: str) -> Any: return datetime.fromisoformat(v) elif _is_int_string(v): return int(v) - elif v.startswith("{") and v.endswith("}") and ":" in v and "{" not in v[1:-1] and "}" not in v[ - 1:-1]: # nested object + elif ( + v.startswith("{") + and v.endswith("}") + and ":" in v + and "{" not in v[1:-1] + and "}" not in v[1:-1] + ): # nested object # check if it is a json dict try: json.loads(v) diff --git a/amqp/adapter/serializer.py b/amqp/adapter/serializer.py index 1543802..5b1619c 100644 --- a/amqp/adapter/serializer.py +++ b/amqp/adapter/serializer.py @@ -8,16 +8,16 @@ from typing import List, Union, Dict def long_to_bytes(x: int) -> bytes: - return struct.pack('q', x) + return struct.pack("q", x) def bytes_to_long(b: bytes) -> int: - return struct.unpack('q', b)[0] + return struct.unpack("q", b)[0] def read_long(bis: BytesIO) -> int: try: - return struct.unpack('q', bis.read(8))[0] + return struct.unpack("q", bis.read(8))[0] except IOError: return 0 @@ -29,37 +29,37 @@ def object_or_list_as_string(value: Union[str, List[str]]) -> str: def map_as_string(map_: Dict[str, str]) -> str: - return ','.join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items()) + return ",".join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items()) def map_with_list_as_string(map_: Dict[str, List[str]]) -> str: - return ','.join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items()) + return ",".join(f"{k}={object_or_list_as_string(v)}" for k, v in map_.items()) def parse_map_string(input_str: str) -> Dict[str, str]: map_ = {} - if input_str.startswith('{') and input_str.endswith('}'): - for entry in input_str[1:-1].split(','): + if input_str.startswith("{") and input_str.endswith("}"): + for entry in input_str[1:-1].split(","): if len(entry) > 1: - key, value = entry.split('=') + key, value = entry.split("=") map_[key] = value return map_ def csv_as_list(input_str: str) -> List[str]: - return [item.strip() for item in input_str.strip('[]').split(',')] + return [item.strip() for item in input_str.strip("[]").split(",")] def add_to_map(map_: Dict[str, List[str]], token: str) -> None: - eq_idx = token.index('=') + eq_idx = token.index("=") if eq_idx > 0: key = token[:eq_idx] - map_[key] = csv_as_list(token[eq_idx+1:]) + map_[key] = csv_as_list(token[eq_idx + 1 :]) def parse_map_list_string(input_str: str) -> Dict[str, List[str]]: map_ = {} - if input_str.startswith('{') and input_str.endswith('}'): - for token in input_str[1:-1].split('],'): + if input_str.startswith("{") and input_str.endswith("}"): + for token in input_str[1:-1].split("],"): add_to_map(map_, token) return map_ diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py index 4f62a92..3ac1910 100644 --- a/amqp/adapter/service_message_factory.py +++ b/amqp/adapter/service_message_factory.py @@ -20,6 +20,7 @@ class ServiceMessageFactory: """ build/serialize/deserialize CleverMicro Service Message class """ + def __init__(self, name: str): self.process_name = name @@ -37,7 +38,9 @@ class ServiceMessageFactory: ) return "null" - def of(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> ServiceMessage: + def of( + self, message_type: ServiceMessageType, payload: str, recipient_name: str + ) -> ServiceMessage: """ Builds ServiceMessage object from relevant values. :param message_type: enum Message type @@ -53,10 +56,12 @@ class ServiceMessageFactory: recipient_name=recipient_name, reply_to=self.process_name, trace_info=TraceInfoAdapter.create_produce_span(_id), - message=payload + message=payload, ) - def as_base64(self, message_type: ServiceMessageType, payload: str, recipient_name: str) -> ServiceMessage: + def as_base64( + self, message_type: ServiceMessageType, payload: str, recipient_name: str + ) -> ServiceMessage: """ Builds ServiceMessage object from relevant values, stores content Base64 encoded. :param message_type: enum Message type @@ -72,7 +77,7 @@ class ServiceMessageFactory: recipient_name=recipient_name, reply_to=self.process_name, trace_info=TraceInfoAdapter.create_produce_span(_id.as_string()), - message=payload + message=payload, ) def from_bytes(self, input_bytes: bytes) -> ServiceMessage: @@ -94,7 +99,9 @@ class ServiceMessageFactory: i += 1 reply_to = input_str_array[i] if i < len(input_str_array) else "" i += 1 - trace_info = parse_map_string(input_str_array[i]) if i < len(input_str_array) else {} + trace_info = ( + parse_map_string(input_str_array[i]) if i < len(input_str_array) else {} + ) i += 1 message = input_str_array[i] if i < len(input_str_array) else "" return ServiceMessage( @@ -104,14 +111,20 @@ class ServiceMessageFactory: recipient_name=recipient_name, reply_to=reply_to, trace_info=trace_info, - message=base64.b64decode(message).decode() if payload_type & 0x80 else message + message=( + base64.b64decode(message).decode() + if payload_type & 0x80 + else message + ), ) except Exception as e: logging_error( "Unexpected ServiceMessage format. Expected 6 tokens separated by '|' in format: " "[ID(64bit number)|MessageType(8bit number)|RecipientName(String)|" "ReplyTo(String)|TraceInfo|Payload(String or Base64)], got: " - "[%s], reported problem: %s", input_bytes.decode(), str(e) + "[%s], reported problem: %s", + input_bytes.decode(), + str(e), ) return INVALID_MESSAGE @@ -125,7 +138,7 @@ class ServiceMessageFactory: f"{message.id.as_string()}{RS}{self.format_message_type(message)}{RS}{message.recipient_name}{RS}" f"{message.reply_to}{RS}{map_as_string(message.trace_info)}{RS}" f"{base64.b64encode(message.message.encode()).decode() if message.payload_type != 0 else message.message}" - ).encode('utf-8') + ).encode("utf-8") def format_message_type(self, message: ServiceMessage) -> str: """ @@ -134,6 +147,10 @@ class ServiceMessageFactory: :param message: Service message to send :return: formatted message type """ - num_type = (message.message_type.value if message.message_type else ServiceMessageType.UNKNOWN.value) - 1 - fmt = ((num_type & 0x7F) | (message.payload_type & 0x80)) + num_type = ( + message.message_type.value + if message.message_type + else ServiceMessageType.UNKNOWN.value + ) - 1 + fmt = (num_type & 0x7F) | (message.payload_type & 0x80) return f"{fmt:02X}" diff --git a/amqp/adapter/test_amq_route_factory.py b/amqp/adapter/test_amq_route_factory.py index c7c9aac..495e094 100644 --- a/amqp/adapter/test_amq_route_factory.py +++ b/amqp/adapter/test_amq_route_factory.py @@ -7,14 +7,24 @@ from amqp.model.model import AMQRoute class TestAMQRouteFactory(TestCase): def test_from_string(self): _f = AMQRouteFactory() - _r: AMQRoute = _f.from_string("amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0") + _r: AMQRoute = _f.from_string( + "amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0" + ) self.assertEqual("", _r.exchange) self.assertEqual("amq-clever-pybook", _r.queue) self.assertEqual(5, _r.timeout) - _r2: AMQRoute = _f.from_string("amq-clever-pybook:amq-clever-pybook:clever.pybook.localhost.#:5") + _r2: AMQRoute = _f.from_string( + "amq-clever-pybook:amq-clever-pybook:clever.pybook.localhost.#:5" + ) self.assertEqual(NULL_ROUTE, _r2) def test_validate(self): _f = AMQRouteFactory() - self.assertTrue(_f.validate("amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0")) - self.assertFalse(_f.validate("amq-clever-pybook:amq-clever-pybook:clever.pybook.localhost.#")) + self.assertTrue( + _f.validate( + "amq-clever-pybook::amq-clever-pybook:clever.pybook.localhost.#:5:0" + ) + ) + self.assertFalse( + _f.validate("amq-clever-pybook:amq-clever-pybook:clever.pybook.localhost.#") + ) diff --git a/amqp/adapter/test_data_message_factory.py b/amqp/adapter/test_data_message_factory.py index d430bf3..75e0e17 100644 --- a/amqp/adapter/test_data_message_factory.py +++ b/amqp/adapter/test_data_message_factory.py @@ -9,37 +9,76 @@ class DataMessageFactoryTest(TestCase): raw = "\002\027A64f3b5af4d00000a4000|m=POST|d=clever-pybook.localhost|p=/debug|h={Accept=[*/*],X-Forwarded-Host=[clever-pybook.localhost:8085],User-Agent=[curl/8.7.1],X-Forwarded-Proto=[http],X-Forwarded-For=[172.18.0.1],Host=[clever-pybook.localhost:8085],Accept-Encoding=[gzip],Content-Length=[1475],X-Forwarded-Port=[8085],X-Real-Ip=[172.18.0.1],X-Forwarded-Server=[c5945bef67d1],Content-Type=[multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v]}|t={traceparent=00-bed6eb7de0d8ccf4c80b145a4bfec357-8d1f3bafc471bf0f-01}|b=TEST_ME" msg = DataMessageFactory.from_bytes(raw.encode("utf-8")) self.assertIsNotNone(msg.id) - self.assertEqual(msg.id.as_string(),'64F3B5AF4D00000A4000'.lower(),'SnowflakeID incorrectly parsed.') - self.assertEqual(msg.headers["Content-Type"][0], "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v") + self.assertEqual( + msg.id.as_string(), + "64F3B5AF4D00000A4000".lower(), + "SnowflakeID incorrectly parsed.", + ) + self.assertEqual( + msg.headers["Content-Type"][0], + "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v", + ) def test_generate_snowflake_id(self): _id = DataMessageFactory.get_instance(1).generate_snowflake_id() - self.assertEqual('00001000', _id.as_string()[-8:], 'Instance ID not set correctly') + self.assertEqual( + "00001000", _id.as_string()[-8:], "Instance ID not set correctly" + ) _id = DataMessageFactory(2).generate_snowflake_id() - self.assertEqual('00002000', _id.as_string()[-8:], 'Instance ID not set correctly') + self.assertEqual( + "00002000", _id.as_string()[-8:], "Instance ID not set correctly" + ) _f = DataMessageFactory(127) _i0 = _f.generate_snowflake_id() _i1 = _f.generate_snowflake_id() _i2 = _f.generate_snowflake_id() _i3 = _f.generate_snowflake_id() - self.assertEqual('0007F000'.lower(), _i0.as_string()[-8:], 'Instance ID and sequence not set correctly') - self.assertEqual('0007F001'.lower(), _i1.as_string()[-8:], 'Instance ID and sequence not set correctly') - self.assertEqual('0007F002'.lower(), _i2.as_string()[-8:], 'Instance ID and sequence not set correctly') - self.assertEqual('0007F003'.lower(), _i3.as_string()[-8:], 'Instance ID and sequence not set correctly') + self.assertEqual( + "0007F000".lower(), + _i0.as_string()[-8:], + "Instance ID and sequence not set correctly", + ) + self.assertEqual( + "0007F001".lower(), + _i1.as_string()[-8:], + "Instance ID and sequence not set correctly", + ) + self.assertEqual( + "0007F002".lower(), + _i2.as_string()[-8:], + "Instance ID and sequence not set correctly", + ) + self.assertEqual( + "0007F003".lower(), + _i3.as_string()[-8:], + "Instance ID and sequence not set correctly", + ) def test_create_request_message(self): _f = DataMessageFactory(34) - _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox") - self.assertEqual("localhost", _req.domain, 'Domain not set correctly') - self.assertEqual("/test/me", _req.path, 'Context path not set correctly') - self.assertEqual("v1", _req.headers['k1'][0], "Headers not set correctly") - self.assertEqual(b'Quick brown fox', _req.body(), "Body not correctly decoded") + _req = _f.create_request_message( + "PUT", + "localhost", + "/test/me", + {"k1": ["v1"], "k2": ["v2"]}, + "Quick brown fox", + ) + self.assertEqual("localhost", _req.domain, "Domain not set correctly") + self.assertEqual("/test/me", _req.path, "Context path not set correctly") + self.assertEqual("v1", _req.headers["k1"][0], "Headers not set correctly") + self.assertEqual(b"Quick brown fox", _req.body(), "Body not correctly decoded") def test_serialize(self): _f = DataMessageFactory(34) - _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox") + _req = _f.create_request_message( + "PUT", + "localhost", + "/test/me", + {"k1": ["v1"], "k2": ["v2"]}, + "Quick brown fox", + ) _ser = _f.serialize(_req) self.assertEqual(65, _ser[2]) _deser = _f.from_bytes(_ser) @@ -51,18 +90,40 @@ class DataMessageFactoryTest(TestCase): def test_amq_message_routing_key(self): _f = DataMessageFactory(35) - _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox") + _req = _f.create_request_message( + "PUT", + "localhost", + "/test/me", + {"k1": ["v1"], "k2": ["v2"]}, + "Quick brown fox", + ) _key = _f.amq_message_routing_key(_req) - self.assertEqual("localhost.test.me", _key, "Message routing key wrongly extracted from the message") + self.assertEqual( + "localhost.test.me", + _key, + "Message routing key wrongly extracted from the message", + ) def test_get_timestamp_from_id(self): _f = DataMessageFactory(35) _timestamp = _f.get_timestamp_from_id(_f.generate_snowflake_id()) _x_timestamp = int(datetime.now(timezone.utc).timestamp() * 1000) - self.assertLessEqual(_x_timestamp - _timestamp, 1, "Timestamp in SnowflakeId differs too much from current time") + self.assertLessEqual( + _x_timestamp - _timestamp, + 1, + "Timestamp in SnowflakeId differs too much from current time", + ) def test_to_string(self): _f = DataMessageFactory(35) - _req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox") + _req = _f.create_request_message( + "PUT", + "localhost", + "/test/me", + {"k1": ["v1"], "k2": ["v2"]}, + "Quick brown fox", + ) _as_str = _f.to_string(_req) - self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found") + self.assertGreater( + _as_str.index("domain='localhost'"), 0, "Expected content not found" + ) diff --git a/amqp/adapter/test_data_response_factory.py b/amqp/adapter/test_data_response_factory.py index a6e5c83..68f758e 100644 --- a/amqp/adapter/test_data_response_factory.py +++ b/amqp/adapter/test_data_response_factory.py @@ -20,9 +20,13 @@ class TestDataResponseFactory(TestCase): _snowflakeId: SnowflakeId = _g.generate_snowflake_id() _resp = _f.create_async_response_message(_snowflakeId) _ser = _f.serialize(_resp) - self.assertEqual(2, str(_ser[2:]).index(_f.MAGIC_BYTE_HTTP_REQUEST+_snowflakeId.as_string()), "Magic byte and SnowflakeId not found") + self.assertEqual( + 2, + str(_ser[2:]).index(_f.MAGIC_BYTE_HTTP_REQUEST + _snowflakeId.as_string()), + "Magic byte and SnowflakeId not found", + ) self.assertGreater(str(_ser[2:]).index("|e=|"), 2) - self.assertEqual((_ser[0]+_ser[1]), len(_ser) - 2 - len('T0s=')) + self.assertEqual((_ser[0] + _ser[1]), len(_ser) - 2 - len("T0s=")) def test_from_bytes(self): _f = DataResponseFactory() @@ -33,4 +37,4 @@ class TestDataResponseFactory(TestCase): _deser = _f.from_bytes(_ser) self.assertEqual(_snowflakeId.as_string(), _deser.id) self.assertEqual(200, _deser.response_code) - self.assertEqual(b'OK', _deser.response) + self.assertEqual(b"OK", _deser.response) diff --git a/amqp/adapter/test_file_downloader.py b/amqp/adapter/test_file_downloader.py index de24c72..fe18423 100644 --- a/amqp/adapter/test_file_downloader.py +++ b/amqp/adapter/test_file_downloader.py @@ -6,6 +6,7 @@ import pytest from amqp.adapter.file_downloader import ensure_directory_exists + class TestFileDownloader(TestCase): def test_ensure_directory_exists(self): """ @@ -30,7 +31,9 @@ class TestFileDownloader(TestCase): with open(filepath2, "w") as f: f.write("test") result2_v1 = ensure_directory_exists(filepath2) - assert result2_v1 == os.path.join(temp_dir, "test2.1.txt"), f"Test Case 2 Version 1 Failed: {result2_v1} != {os.path.join(temp_dir, 'test2.1.txt')}" + assert result2_v1 == os.path.join( + temp_dir, "test2.1.txt" + ), f"Test Case 2 Version 1 Failed: {result2_v1} != {os.path.join(temp_dir, 'test2.1.txt')}" # Create a few versioned files with open(os.path.join(temp_dir, "test3.txt"), "w") as f: @@ -41,6 +44,8 @@ class TestFileDownloader(TestCase): f.write("test") filepath3 = os.path.join(temp_dir, "test3.txt") result3_v3 = ensure_directory_exists(filepath3) - assert result3_v3 == os.path.join(temp_dir, "test3.3.txt"), f"Test Case 3 Version 3 Failed: {result3_v3} != {os.path.join(temp_dir, 'test3.3.txt')}" + assert result3_v3 == os.path.join( + temp_dir, "test3.3.txt" + ), f"Test Case 3 Version 3 Failed: {result3_v3} != {os.path.join(temp_dir, 'test3.3.txt')}" print("All test cases passed!") diff --git a/amqp/adapter/test_pydantic_serializer.py b/amqp/adapter/test_pydantic_serializer.py index 2e56136..475f265 100644 --- a/amqp/adapter/test_pydantic_serializer.py +++ b/amqp/adapter/test_pydantic_serializer.py @@ -28,9 +28,13 @@ class PydanticSerializerTest(TestCase): assert False def test_parse_url_query(self): - path, args = parse_url_query("http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741?response_type=JsonFile") + path, args = parse_url_query( + "http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741?response_type=JsonFile" + ) assert path == "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741" - assert args == {'response_type': 'JsonFile'} - path, args = parse_url_query("http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741") + assert args == {"response_type": "JsonFile"} + path, args = parse_url_query( + "http://localhost:8080/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741" + ) assert path == "/api/v0/unstructured/863e4d9f-4612-4173-8d0a-3de34cc7a741" assert args == {} diff --git a/amqp/adapter/test_service_message_factory.py b/amqp/adapter/test_service_message_factory.py index 3dfaa8e..72ab467 100644 --- a/amqp/adapter/test_service_message_factory.py +++ b/amqp/adapter/test_service_message_factory.py @@ -1,6 +1,10 @@ from unittest import TestCase -from amqp.adapter.service_message_factory import ServiceMessageFactory, RS, INVALID_MESSAGE +from amqp.adapter.service_message_factory import ( + ServiceMessageFactory, + RS, + INVALID_MESSAGE, +) from amqp.model.model import ServiceMessage from amqp.model.service_message_type import ServiceMessageType @@ -8,17 +12,21 @@ from amqp.model.service_message_type import ServiceMessageType class ServiceMessageTest(TestCase): def test_to_string(self): - _f: ServiceMessageFactory = ServiceMessageFactory('bumble-bee') - message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") + _f: ServiceMessageFactory = ServiceMessageFactory("bumble-bee") + message: ServiceMessage = _f.of( + ServiceMessageType.ROUTING_DATA, "Duck", "Donald" + ) message_str = _f.to_string(message) self.assertTrue("|type=01|" in message_str) self.assertTrue("|body=Duck" in message_str) - message: ServiceMessage = _f.of(ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald") + message: ServiceMessage = _f.of( + ServiceMessageType.ROUTING_DATA_REPLACE, "Duck", "Donald" + ) message_str = _f.to_string(message) self.assertTrue("|type=03|" in message_str) self.assertTrue("|body=Duck" in message_str) self.assertTrue("|replyTo=bumble-bee" in message_str) - nullMsg = _f.to_string(None) + nullMsg = _f.to_string(None) self.assertEqual("null", nullMsg) def test_from(self): @@ -31,7 +39,7 @@ class ServiceMessageTest(TestCase): assert deserialized.recipient_name == "Donald" assert deserialized.message == "Duck" assert deserialized.reply_to == "amqp-router" - invalidMsg = _f.from_bytes(b'wrong') + invalidMsg = _f.from_bytes(b"wrong") self.assertEqual(invalidMsg, INVALID_MESSAGE) def test_base64(self): @@ -43,7 +51,7 @@ class ServiceMessageTest(TestCase): assert deserialized.is_valid() assert deserialized.message_type == ServiceMessageType.ROUTING_DATA assert deserialized.message == "Duck" - + def test_serialize(self): _f: ServiceMessageFactory = ServiceMessageFactory("amqp-router") message = _f.of(ServiceMessageType.ROUTING_DATA, "Duck", "Donald") diff --git a/amqp/adapter/trace_info_adapter.py b/amqp/adapter/trace_info_adapter.py index b152bd8..4a87c98 100644 --- a/amqp/adapter/trace_info_adapter.py +++ b/amqp/adapter/trace_info_adapter.py @@ -1,6 +1,7 @@ from dataclasses import dataclass from typing import Dict + @dataclass class TraceInfoAdapter: """ @@ -9,6 +10,4 @@ class TraceInfoAdapter: @staticmethod def create_produce_span(message_id: str) -> Dict[str, str]: - return { - "trace-id": message_id - } + return {"trace-id": message_id} diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py index aeed27e..31bbfab 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -14,7 +14,7 @@ class AMQAdapter: configuration with prefix 'cm.amq-adapter' """ - PREFIX: str = 'cm.amq-adapter.' + PREFIX: str = "cm.amq-adapter." def __init__(self, config: configparser.ConfigParser): """ @@ -24,12 +24,21 @@ class AMQAdapter: :param config: config parser to read configuration values """ - self.generator_id = config.getint('CleverMicro-AMQ', self.PREFIX + 'generator-id', fallback=164) - self.service_name = config.get('CleverMicro-AMQ', self.PREFIX + 'service-name', - fallback='amq-python-adapter-' + str(self.generator_id)) - self.route_mapping = config.get('CleverMicro-AMQ', self.PREFIX + 'route-mapping', fallback=None) + self.generator_id = config.getint( + "CleverMicro-AMQ", self.PREFIX + "generator-id", fallback=164 + ) + self.service_name = config.get( + "CleverMicro-AMQ", + self.PREFIX + "service-name", + fallback="amq-python-adapter-" + str(self.generator_id), + ) + self.route_mapping = config.get( + "CleverMicro-AMQ", self.PREFIX + "route-mapping", fallback=None + ) self.require_authenticated_user = config.get( - 'CleverMicro-AMQ', self.PREFIX + 'require-authenticated-user', fallback=False + "CleverMicro-AMQ", + self.PREFIX + "require-authenticated-user", + fallback=False, ) def adapter_prefix(self) -> str: @@ -41,7 +50,7 @@ class Dispatch: configuration with prefix 'cm.dispatch' """ - PREFIX: str = 'cm.dispatch.' + PREFIX: str = "cm.dispatch." def __init__(self, config: configparser.ConfigParser): """ @@ -55,24 +64,43 @@ class Dispatch: :param config: config parser to read configuration values """ - self.use_dlq = config.getboolean('CleverMicro-AMQ', self.PREFIX + 'use-dlq', fallback=False) - self.use_confirms = config.getboolean('CleverMicro-AMQ', self.PREFIX + 'use-confirms', fallback=False) - self.amq_host = config.get('CleverMicro-AMQ', self.PREFIX + 'amq-host', fallback='rabbitmq') - self.amq_port = config.getint('CleverMicro-AMQ', self.PREFIX + 'amq-port', fallback=5672) - self.amq_port_tls = config.getint('CleverMicro-AMQ', self.PREFIX + 'amq-port-tls', fallback=5671) - self.service_host = config.get('CleverMicro-AMQ', self.PREFIX + 'service-host', fallback='localhost') - self.service_port = config.getint('CleverMicro-AMQ', self.PREFIX + 'service-port', fallback=8080) + self.use_dlq = config.getboolean( + "CleverMicro-AMQ", self.PREFIX + "use-dlq", fallback=False + ) + self.use_confirms = config.getboolean( + "CleverMicro-AMQ", self.PREFIX + "use-confirms", fallback=False + ) + self.amq_host = config.get( + "CleverMicro-AMQ", self.PREFIX + "amq-host", fallback="rabbitmq" + ) + self.amq_port = config.getint( + "CleverMicro-AMQ", self.PREFIX + "amq-port", fallback=5672 + ) + self.amq_port_tls = config.getint( + "CleverMicro-AMQ", self.PREFIX + "amq-port-tls", fallback=5671 + ) + self.service_host = config.get( + "CleverMicro-AMQ", self.PREFIX + "service-host", fallback="localhost" + ) + self.service_port = config.getint( + "CleverMicro-AMQ", self.PREFIX + "service-port", fallback=8080 + ) self.rabbit_mq_user = os.getenv("RABBIT_MQ_USER", "springuser") self.rabbit_mq_password = os.getenv("RABBIT_MQ_PASSWORD", "TheCleverWho") - self.rabbit_mq_url = f"amqp://{self.rabbit_mq_user}:{self.rabbit_mq_password}@{self.amq_host}/" - self.download_dir = config.get('CleverMicro-AMQ', self.PREFIX + 'download-dir', fallback='downloaded_files') + self.rabbit_mq_url = ( + f"amqp://{self.rabbit_mq_user}:{self.rabbit_mq_password}@{self.amq_host}/" + ) + self.download_dir = config.get( + "CleverMicro-AMQ", self.PREFIX + "download-dir", fallback="downloaded_files" + ) + class Backpressure: """ configuration with prefix 'cm.backpressure' """ - PREFIX: str = 'cm.backpressure.' + PREFIX: str = "cm.backpressure." def __init__(self, config: configparser.ConfigParser): """ @@ -81,15 +109,20 @@ class Backpressure: :param config: config parser to read configuration values """ - self.threshold = config.getint('CleverMicro-AMQ', self.PREFIX + 'threshold', fallback=0) + self.threshold = config.getint( + "CleverMicro-AMQ", self.PREFIX + "threshold", fallback=0 + ) # time-window is in milliseconds, to report backpressure IDLE alert - self.time_window = config.getint('CleverMicro-AMQ', self.PREFIX + 'time-window', fallback=10000) + self.time_window = config.getint( + "CleverMicro-AMQ", self.PREFIX + "time-window", fallback=10000 + ) class AMQConfiguration: """ Top level configuration context holder / object """ + def __init__(self, config_file_name: str = None): config = configparser.ConfigParser() # Read the application.properties file @@ -97,7 +130,9 @@ class AMQConfiguration: config.read(config_file_name) else: logging_error(f" [E] Configuration file not found: {config_file_name}") - logging.warning(" [W] Defaulting to preset values, which will likely be wrong, causing errors.") + logging.warning( + " [W] Defaulting to preset values, which will likely be wrong, causing errors." + ) # Override values with environment variables for section in config.sections(): diff --git a/amqp/healthcheck.py b/amqp/healthcheck.py index 761bc30..fb86847 100644 --- a/amqp/healthcheck.py +++ b/amqp/healthcheck.py @@ -1,9 +1,9 @@ import sys import requests -url = 'http://localhost:8080/amq-adapter-healthcheck' -headers = {'Authorization': 'Bearer my_access_token'} -params = {'limit': 10, 'offset': 20} +url = "http://localhost:8080/amq-adapter-healthcheck" +headers = {"Authorization": "Bearer my_access_token"} +params = {"limit": 10, "offset": 20} response = requests.get(url, headers=headers, params=params) if response.status_code == 200: sys.exit(0) diff --git a/amqp/model/model.py b/amqp/model/model.py index 6fddd69..200c114 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -19,8 +19,17 @@ Define the structure of messages used in Clever Micro class CleverMicroMessage: DEFAULT_CONTENT_TYPE = ["application/octet-stream"] - def __init__(self, magic: str, id: SnowflakeId, m_field: str, d_field: str, p_field: str, - headers: Dict[str, List[str]], trace_info: Dict[str, str], base64body: bytes): + def __init__( + self, + magic: str, + id: SnowflakeId, + m_field: str, + d_field: str, + p_field: str, + headers: Dict[str, List[str]], + trace_info: Dict[str, str], + base64body: bytes, + ): self._magic = magic self._id: SnowflakeId = id self._m_field = m_field @@ -62,9 +71,20 @@ class CleverMicroMessage: class DataMessage(CleverMicroMessage): - def __init__(self, magic: str, id: SnowflakeId, method: str, domain: str, path: str, - headers: Dict[str, List[str]], trace_info: Dict[str, str], base64body: bytes): - super().__init__(magic, id, method, domain, path, headers, trace_info, base64body) + def __init__( + self, + magic: str, + id: SnowflakeId, + method: str, + domain: str, + path: str, + headers: Dict[str, List[str]], + trace_info: Dict[str, str], + base64body: bytes, + ): + super().__init__( + magic, id, method, domain, path, headers, trace_info, base64body + ) def method(self) -> str: return self.m_field() @@ -77,9 +97,27 @@ class DataMessage(CleverMicroMessage): class DataResponse(CleverMicroMessage): - def __init__(self, magic: str, id: SnowflakeId, response_code: str, error: str, error_cause: str, - headers: Dict[str, List[str]], trace_info: Dict[str, str], base64body: bytes): - super().__init__(magic, id, response_code, error, error_cause, headers, trace_info, base64body) + def __init__( + self, + magic: str, + id: SnowflakeId, + response_code: str, + error: str, + error_cause: str, + headers: Dict[str, List[str]], + trace_info: Dict[str, str], + base64body: bytes, + ): + super().__init__( + magic, + id, + response_code, + error, + error_cause, + headers, + trace_info, + base64body, + ) def response_code(self) -> str: return self.m_field() @@ -91,7 +129,6 @@ class DataResponse(CleverMicroMessage): return self.p_field() - @dataclass(frozen=True) class AMQRoute: """ @@ -105,6 +142,7 @@ class AMQRoute: :param timeout: Timeout in milliseconds, how long to wait for reply, or not waiting if less then 1 :param valid_until: optional timestamp (millis from epoch) after which the route is ignored """ + component_name: str exchange: str queue: str @@ -122,9 +160,11 @@ class AMQRoute: def __eq__(self, other): if isinstance(other, AMQRoute): - return (self.component_name == other.component_name - and self.queue == other.queue - and self.key == other.key) + return ( + self.component_name == other.component_name + and self.queue == other.queue + and self.key == other.key + ) return False @property @@ -147,6 +187,7 @@ class DataMessage: 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 @@ -165,10 +206,16 @@ class DataMessage: class MultipartDataMessage(DataMessage): def __init__(self, message: DataMessage, extra_data: dict): - super().__init__(message.magic, message.id, - message.method, message.domain, message.path, - message.headers, message.trace_info, - message.base64_body) + super().__init__( + message.magic, + message.id, + message.method, + message.domain, + message.path, + message.headers, + message.trace_info, + message.base64_body, + ) self.extra_data = extra_data @@ -178,6 +225,7 @@ 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 @@ -200,6 +248,7 @@ class ServiceMessage: The CleverMicro service message that is used to communicate the internal state of the AMQ channel. In v0.2 it supports service discovery / route registration and Backpressure status notification. """ + id: SnowflakeId = None payload_type: int = 0 message_type: ServiceMessageType = ServiceMessageType.UNKNOWN @@ -229,17 +278,20 @@ class ScalingRequest: def to_json(self) -> str: """Converts the object to a JSON string matching the Java structure""" - return json.dumps({ - "version": self.version, - "serviceId": self.service_id, - "taskId": self.task_id, - "maxAvailability": self.max_availability, - "currentAvailability": self.current_availability, - "requestType": self.request_type.value # Using .value for Enum serialization - }, indent=2) + return json.dumps( + { + "version": self.version, + "serviceId": self.service_id, + "taskId": self.task_id, + "maxAvailability": self.max_availability, + "currentAvailability": self.current_availability, + "requestType": self.request_type.value, # Using .value for Enum serialization + }, + indent=2, + ) @classmethod - def from_json(cls, json_str: str) -> 'ScalingRequest': + def from_json(cls, json_str: str) -> "ScalingRequest": """Creates a ScalingRequest from a JSON string""" data = json.loads(json_str) return cls( @@ -247,5 +299,7 @@ class ScalingRequest: task_id=data["taskId"], max_availability=data["maxAvailability"], current_availability=data["currentAvailability"], - request_type=ScalingRequestAlert(data["requestType"]) # Convert string to Enum + request_type=ScalingRequestAlert( + data["requestType"] + ), # Convert string to Enum ) diff --git a/amqp/model/snowflake_id.py b/amqp/model/snowflake_id.py index 655da64..4567d51 100644 --- a/amqp/model/snowflake_id.py +++ b/amqp/model/snowflake_id.py @@ -5,9 +5,12 @@ class SnowflakeId: """ see https://en.wikipedia.org/wiki/Snowflake_ID """ - SNOWFLAKE_ID_WIDTH = 10 # with in bytes (8 bits each), can be more than long int width (8 bytes) - def __init__(self, hi_bits_or_bytes = None, lo_bits = None): + SNOWFLAKE_ID_WIDTH = ( + 10 # with in bytes (8 bits each), can be more than long int width (8 bytes) + ) + + def __init__(self, hi_bits_or_bytes=None, lo_bits=None): self.bits = [0, 0] if lo_bits is None: if hi_bits_or_bytes is not None: @@ -39,13 +42,17 @@ class SnowflakeId: for i in range(0, expected_length, 2): if i == expected_length - (2 * 8): n = 0 - bits[n] = (bits[n] << 8) | int(_input[i:i + 2], 16) + bits[n] = (bits[n] << 8) | int(_input[i : i + 2], 16) return SnowflakeId(bits[1], bits[0]) def __str__(self): _hex = f"{self.bits[1]:016X}{self.bits[0]:016X}" - _val = _hex if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH):] + _val = ( + _hex + if len(_hex) <= 2 * SnowflakeId.SNOWFLAKE_ID_WIDTH + else _hex[(-2 * SnowflakeId.SNOWFLAKE_ID_WIDTH) :] + ) return _val.lower() def __eq__(self, other): diff --git a/amqp/model/test_snowflake_id.py b/amqp/model/test_snowflake_id.py index 5027d12..c39b786 100644 --- a/amqp/model/test_snowflake_id.py +++ b/amqp/model/test_snowflake_id.py @@ -18,7 +18,6 @@ class SnowflakeIdTest(TestCase): _id_2 = DataMessageFactory.get_instance(1).generate_snowflake_id() self.assertTrue(_id_1 < _id_2) - def test_from_hex(self): _factory = DataMessageFactory.get_instance(1) _id_1 = _factory.generate_snowflake_id() @@ -27,8 +26,10 @@ class SnowflakeIdTest(TestCase): self.assertTrue(_id_1 == _id_2) with self.assertRaises(RuntimeError) as context: SnowflakeId.from_hex("100") - self.assertTrue(f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string " - in str(context.exception)) + self.assertTrue( + f"MessageId is {SnowflakeId.SNOWFLAKE_ID_WIDTH * 8} bits wide, it is represented by HEX string " + in str(context.exception) + ) def test_get_bytes(self): _factory = DataMessageFactory.get_instance(1) @@ -41,7 +42,6 @@ class SnowflakeIdTest(TestCase): _id_1 = _factory.generate_snowflake_id() self.assertEqual(2, len(_id_1.get_bits())) - def test_as_string(self): _factory = DataMessageFactory.get_instance(1) _id_1 = _factory.generate_snowflake_id() diff --git a/amqp/rabbitmq/rabbit_mq_client.py b/amqp/rabbitmq/rabbit_mq_client.py index e1be643..b4dc2fe 100644 --- a/amqp/rabbitmq/rabbit_mq_client.py +++ b/amqp/rabbitmq/rabbit_mq_client.py @@ -7,7 +7,11 @@ 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_base import ( + AMQ_REPLY_TO_EXCHANGE, + DLQ_EXCHANGE, + AMQ_RPC_EXCHANGE, +) from amqp.router.router_consumer import RouterConsumer from amqp.router.utils import sanitize_as_rabbitmq_name @@ -23,15 +27,20 @@ class RabbitMQClient: self.reply_exchange: aio_pika.RobustExchange | None = None self.amq_configuration = configuration logging.info("*******************************************") - logging.info("******** RabbitMQProducer.init(): %s", - self.amq_configuration.dispatch.amq_host) + logging.info( + "******** RabbitMQProducer.init(): %s", + self.amq_configuration.dispatch.amq_host, + ) logging.info("*******************************************") self.router = RouterConsumer(configuration) # this.meterRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT); - - self.reply_to_exchange: AbstractRobustExchange | None = None # when sending response to RPC request - self.reply_to_queue: AbstractRobustQueue | None= None # when receiving response to RPC request + self.reply_to_exchange: AbstractRobustExchange | None = ( + None # when sending response to RPC request + ) + self.reply_to_queue: AbstractRobustQueue | None = ( + None # when receiving response to RPC request + ) async def init(self, loop) -> AbstractRobustExchange: await self.setup_amq_channel(loop) @@ -43,7 +52,9 @@ class RabbitMQClient: self.reply_to_exchange = await self.channel.declare_exchange( AMQ_REPLY_TO_EXCHANGE, type=ExchangeType.direct ) - await self.reply_to_queue.bind(exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue) + await self.reply_to_queue.bind( + exchange=self.reply_to_exchange, routing_key=cm_reply_to_queue + ) return self.reply_to_exchange def cleanup(self): @@ -55,7 +66,10 @@ 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): """ @@ -66,36 +80,58 @@ class RabbitMQClient: After this the Service is listening on (consuming from) the queue for incoming messages. """ if self.router.is_open(self.channel): - queue_args = self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None - logging.info(" [*] RabbitMQClient register routes, cfg mapping: [%s]", route_mapping) + queue_args = ( + self.DLQ_ARGS if self.amq_configuration.dispatch.use_dlq else None + ) + logging.info( + " [*] RabbitMQClient register routes, cfg mapping: [%s]", route_mapping + ) # convert the comma-separated string into list of AMQRoute objects requested_routes = self.router.unwrap_route_list(route_mapping) for route in requested_routes: try: # Note: AMQRoute allows one of the exchange or queue to be None, in which case the default is used - queue_name = route.queue if route.queue else self.router.get_consume_queue_name() + queue_name = ( + route.queue + if route.queue + else self.router.get_consume_queue_name() + ) logging.info(" [*] RabbitMQClient register route: [%s]", route) - queue: AbstractRobustQueue = await self.channel.declare_queue(name=queue_name, - durable=True, - exclusive=False, - auto_delete=False, - arguments=queue_args) - _exchange_name: str = route.exchange if route.exchange else AMQ_RPC_EXCHANGE - exchange: AbstractRobustExchange = await self.channel.declare_exchange( - name=_exchange_name, type=ExchangeType.topic + queue: AbstractRobustQueue = await self.channel.declare_queue( + name=queue_name, + durable=True, + exclusive=False, + auto_delete=False, + arguments=queue_args, + ) + _exchange_name: str = ( + route.exchange if route.exchange else AMQ_RPC_EXCHANGE + ) + exchange: AbstractRobustExchange = ( + await self.channel.declare_exchange( + name=_exchange_name, type=ExchangeType.topic + ) ) await queue.bind(exchange, route.key) _c_tag = await queue.consume(callback, no_ack=False) # Register the route with the router and publish its detail to other adapters await self.router.register_route( AMQRoute( - route.component_name, _exchange_name, queue_name, + route.component_name, + _exchange_name, + queue_name, sanitize_as_rabbitmq_name(route.key), - route.timeout, route.valid_until + route.timeout, + route.valid_until, ) ) - logging.info(" [%s] AMQ connection initialized for route: %s, listens on: %s/%s", - _c_tag, route, exchange, queue_name) + logging.info( + " [%s] AMQ connection initialized for route: %s, listens on: %s/%s", + _c_tag, + route, + exchange, + queue_name, + ) except Exception as err: logging_error("Callback registration INCOMPLETE, %s", err) else: @@ -103,7 +139,11 @@ class RabbitMQClient: async def register_data_reply_callback(self, rpc_callback): _c_tag = await self.reply_to_queue.consume(callback=rpc_callback, no_ack=False) - logging.info(" [%s] RabbitMQClient register data reply callback on queue: %s", _c_tag,self.reply_to_queue.name) + logging.info( + " [%s] RabbitMQClient register data reply callback on queue: %s", + _c_tag, + self.reply_to_queue.name, + ) async def setup_amq_channel(self, loop): """ @@ -115,15 +155,21 @@ class RabbitMQClient: port=self.amq_configuration.dispatch.amq_port, login=self.amq_configuration.dispatch.rabbit_mq_user, password=self.amq_configuration.dispatch.rabbit_mq_password, - loop=loop + loop=loop, + ) + logging.info( + "RabbitMQProducer.setupAMQ, got connection, open=%s", + self.connection.is_closed, ) - logging.info("RabbitMQProducer.setupAMQ, got connection, open=%s", self.connection.is_closed) self.channel = await self.connection.channel() await self.channel.set_qos(prefetch_count=self.PREFETCH_COUNT) await self.router.init_internal_routing_paths(self.channel) self.router.set_route_added_notifier( - lambda route: logging.info(" [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s", - route.exchange)) + lambda route: logging.info( + " [DBG] ---------- RouteAdded, setting up metrics: amqp.adapter.exchange.%s", + route.exchange, + ) + ) async def send_message(self, message: DataMessage): """ @@ -140,31 +186,42 @@ class RabbitMQClient: # No response expected pika_message: aio_pika.message.Message = aio_pika.message.Message( body=DataMessageFactory.serialize(message), - content_encoding='utf-8', - delivery_mode=2 + content_encoding="utf-8", + delivery_mode=2, ) await self.rpc_exchange.publish( message=pika_message, - routing_key=self.router.get_routing_key(message), # context path is a routing key + routing_key=self.router.get_routing_key( + message + ), # context path is a routing key + ) + logging.info( + f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}" ) - logging.info(f" [x] basicPublish (no-reply), messageId: {message.id}, to: {route}") else: # This is RPC call, there should be response pika_message: aio_pika.message.Message = aio_pika.message.Message( body=DataMessageFactory.serialize(message), - content_encoding='utf-8', + content_encoding="utf-8", delivery_mode=2, correlation_id=str(message.id), - reply_to=self.router.get_reply_to_queue_name() + reply_to=self.router.get_reply_to_queue_name(), ) await self.rpc_exchange.publish( message=pika_message, - routing_key=self.router.get_routing_key(message), # context path is a routing key + routing_key=self.router.get_routing_key( + message + ), # context path is a routing key + ) + logging.info( + " [x] basicPublish (RPC call), messageId: %s, to: %s", + message.id.as_string(), + route.as_string(), ) - logging.info(" [x] basicPublish (RPC call), messageId: %s, to: %s", - message.id.as_string(), route.as_string()) else: - logging.warning("Message not sent, no route for %s/%s", message.domain, message.path) + logging.warning( + "Message not sent, no route for %s/%s", message.domain, message.path + ) except Exception as err: logging_error("Send message failed: %s", err) diff --git a/amqp/router/route_database.py b/amqp/router/route_database.py index d7508b1..0e0e77b 100644 --- a/amqp/router/route_database.py +++ b/amqp/router/route_database.py @@ -15,7 +15,7 @@ class RouteDatabase: def pattern(self, routing_key: str) -> re.Pattern: breaker = False - tokens = routing_key.split('.') + tokens = routing_key.split(".") counter = len(tokens) regex_parts = [] @@ -40,9 +40,13 @@ class RouteDatabase: return re.compile(".*") return re.compile(regex) - def matches(self, routing_key_from_route: str, routing_key_from_message: str) -> bool: + def matches( + self, routing_key_from_route: str, routing_key_from_message: str + ) -> bool: route_pattern = self.pattern(routing_key_from_route) - return bool(route_pattern.match(sanitize_as_rabbitmq_name(routing_key_from_message))) + return bool( + route_pattern.match(sanitize_as_rabbitmq_name(routing_key_from_message)) + ) def wrap_routes(self) -> str: return ",".join(str(route) for route in self.routes) diff --git a/amqp/router/router_base.py b/amqp/router/router_base.py index 0a7250e..b98675c 100644 --- a/amqp/router/router_base.py +++ b/amqp/router/router_base.py @@ -38,7 +38,10 @@ class RouterBase: def unwrap_route_list(self, route_string: str) -> List[AMQRoute]: logging.info(" [DBG] RouterMaster.unwrapRoutes, route(s): %s", route_string) - routes = [AMQRouteFactory.from_string(route_str) for route_str in route_string.split(",")] + routes = [ + AMQRouteFactory.from_string(route_str) + for route_str in route_string.split(",") + ] return [route for route in routes if route != NULL_ROUTE] def get_routes(self) -> Set[AMQRoute]: @@ -53,13 +56,19 @@ class RouterBase: def add_routes(self, message: str) -> None: try: amq_route_list = self.unwrap_route_list(message) - logging.info(" [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message) + logging.info( + " [DBG] RouterMaster.addRoute [%d]: %s", len(amq_route_list), message + ) for route in amq_route_list: logging.info(" [*] RouterMaster.addRoute, adding route: %s", route) if not route.exchange and not route.queue and not route.key: - logging_error("Cannot setup AMQ Exchange for '%s', at least one of " - "Exchange, Queue and/or Key must be present in the routing string. " - "Expected format is '%s'", route, AMQRoute.FORMAT) + logging_error( + "Cannot setup AMQ Exchange for '%s', at least one of " + "Exchange, Queue and/or Key must be present in the routing string. " + "Expected format is '%s'", + route, + AMQRoute.FORMAT, + ) else: if route.exchange != DLQ_EXCHANGE: is_new_route = self.route_database.add_route(route) @@ -73,9 +82,14 @@ class RouterBase: def remove_routes(self, message: str) -> None: try: - to_remove = [AMQRouteFactory.from_string(route_str) for route_str in message.split(",")] + to_remove = [ + AMQRouteFactory.from_string(route_str) + for route_str in message.split(",") + ] removed_count = sum(1 for route in to_remove if self.remove_route(route)) - logging.info(f" [*] RouterMaster.removeRoutes(): {message}, removed={removed_count}") + logging.info( + f" [*] RouterMaster.removeRoutes(): {message}, removed={removed_count}" + ) except IOError as err: logging.info(f" [E] Can't remove route(s): {err}") diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py index 5ae79a1..f1e17ac 100644 --- a/amqp/router/router_consumer.py +++ b/amqp/router/router_consumer.py @@ -1,27 +1,36 @@ import logging from typing import Set -from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractIncomingMessage +from aio_pika.abc import ( + AbstractRobustChannel, + AbstractRobustQueue, + AbstractIncomingMessage, +) from amqp.adapter.amq_route_factory import AMQRouteFactory from amqp.adapter.logging_utils import logging_error from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import ServiceMessage, AMQRoute from amqp.model.service_message_type import ServiceMessageType -from amqp.router.router_base import CM_C_REQ_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, ANY_RECIPIENT +from amqp.router.router_base import ( + CM_C_REQ_QUEUE, + CM_REQUEST_EXCHANGE, + CM_RESPONSE_EXCHANGE, + ANY_RECIPIENT, +) from amqp.router.router_producer import RouterProducer class RouterConsumer(RouterProducer): """ - * Contains Consumer logic. The logic here: - * Sets a listener on CM_C_REQ_QUEUE queue to receive routing data requests - * Upon receipt of routing data request, responds on CM_RESPONSE_EXCHANGE with local routing data. - * It also proactively pushes the local routing data to CM_RESPONSE_EXCHANGE on startup to - * notify the other (excluding self) Producers about its presence. - * - * Consumer is also Producer (Consumer sits alongside the CM Service, and anytime the Service needs - * to communicate with other Service, it can do so via the Adapter). Therefore, the Producer - * capability is also required. + * Contains Consumer logic. The logic here: + * Sets a listener on CM_C_REQ_QUEUE queue to receive routing data requests + * Upon receipt of routing data request, responds on CM_RESPONSE_EXCHANGE with local routing data. + * It also proactively pushes the local routing data to CM_RESPONSE_EXCHANGE on startup to + * notify the other (excluding self) Producers about its presence. + * + * Consumer is also Producer (Consumer sits alongside the CM Service, and anytime the Service needs + * to communicate with other Service, it can do so via the Adapter). Therefore, the Producer + * capability is also required. """ def __init__(self, configuration: AMQConfiguration): @@ -31,94 +40,148 @@ class RouterConsumer(RouterProducer): async def init_routing_paths_consumer(self, _channel: AbstractRobustChannel): """ - * Initialize router according to Consumer mode. + * Initialize router according to Consumer mode. """ - logging.info(" [*] RouterConsumer.initConsumerRouter, channel open: %s", self.is_open(_channel)) + logging.info( + " [*] RouterConsumer.initConsumerRouter, channel open: %s", + self.is_open(_channel), + ) if self.is_open(_channel): try: # 1. declare a Queue to receive the RoutingData requests - _cm_request_queue: str = self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_REQ_QUEUE - _queue: AbstractRobustQueue = await self.channel.declare_queue(name=_cm_request_queue, durable=True, exclusive=False, auto_delete=False) + _cm_request_queue: str = ( + self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_REQ_QUEUE + ) + _queue: AbstractRobustQueue = await self.channel.declare_queue( + name=_cm_request_queue, + durable=True, + exclusive=False, + auto_delete=False, + ) # 1a. bind queue to CM_REQUEST_EXCHANGE w/o any conditions (empty routing key) await _queue.bind(exchange=CM_REQUEST_EXCHANGE, routing_key="") # 1b. Service adapter mode - listens on Request queue and replies to Response queue - _c_tag = await _queue.consume(callback=self.handle_routing_data_request_message, no_ack=False) - logging.info(" [%s] RouterConsumer listens for RouteDataRequests on: %s", _c_tag, _cm_request_queue) + _c_tag = await _queue.consume( + callback=self.handle_routing_data_request_message, no_ack=False + ) + logging.info( + " [%s] RouterConsumer listens for RouteDataRequests on: %s", + _c_tag, + _cm_request_queue, + ) except Exception as ioe: logging_error("RouterConsumer FATAL: cannot initialize, %s", ioe) else: - logging_error("RouterConsumer FATAL: cannot initialize, RabbitMQ channel is not open.") + logging_error( + "RouterConsumer FATAL: cannot initialize, RabbitMQ channel is not open." + ) """ * ServiceMessageType.ROUTING_DATA_REQUEST handler listening on queue 'CleverMicroRequest' """ - async def handle_routing_data_request_message(self, message: AbstractIncomingMessage): + async def handle_routing_data_request_message( + self, message: AbstractIncomingMessage + ): # some debug statements, TODO - remove for prod release delivery_tag = message.delivery_tag debug = "Delivery.Properties = " + message.properties.__str__() - logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s", - delivery_tag, message.consumer_tag, message.properties, debug) + logging.info( + " [DBG] ******[%s] (ROUTING_DATA_REQ) tag: %s, env: %s, props: %s", + delivery_tag, + message.consumer_tag, + message.properties, + debug, + ) await message.ack(False) - cm_message: ServiceMessage = self.service_message_factory.from_bytes(message.body) + cm_message: ServiceMessage = self.service_message_factory.from_bytes( + message.body + ) route_mapping: str = self.wrap_own_routes() - logging.info(" [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]", - delivery_tag, self.service_message_factory.to_string(cm_message), route_mapping) + logging.info( + " [DBG] ******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]", + delivery_tag, + self.service_message_factory.to_string(cm_message), + route_mapping, + ) # Reply only if self adapter has own routeMapping (e.g. API-gateway-side adapter has none) - if cm_message.is_valid() and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST and route_mapping: + if ( + cm_message.is_valid() + and cm_message.message_type == ServiceMessageType.ROUTING_DATA_REQUEST + and route_mapping + ): if cm_message.reply_to != self.amq_configuration.amq_adapter.service_name: try: service_message: ServiceMessage = self.service_message_factory.of( - ServiceMessageType.ROUTING_DATA, route_mapping, cm_message.reply_to + ServiceMessageType.ROUTING_DATA, + route_mapping, + cm_message.reply_to, + ) + await self.publish_service_message( + service_message, self.cm_response_exchange ) - await self.publish_service_message(service_message, self.cm_response_exchange) except Exception as err: - logging_error("******[%s] (ROUTING_DATA_REQ) Response to queue: %s failed: %s", - delivery_tag, CM_RESPONSE_EXCHANGE, err) + logging_error( + "******[%s] (ROUTING_DATA_REQ) Response to queue: %s failed: %s", + delivery_tag, + CM_RESPONSE_EXCHANGE, + err, + ) else: - logging.info(" [*] ******[%s] (ROUTING_DATA_REQ) [%s] Ignore request originated by ourselves.") + logging.info( + " [*] ******[%s] (ROUTING_DATA_REQ) [%s] Ignore request originated by ourselves." + ) else: - logging_error("******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.", - delivery_tag, str(message.body)) + logging_error( + "******[%s] (ROUTING_DATA_REQ) [%s] produces invalid or no CM msg type.", + delivery_tag, + str(message.body), + ) async def register_route(self, route: AMQRoute): """ - * Called in Consumer (Service Adapter) context. Registers the routing data for self Service - * with the Master router by sending own routing data to Master via CleverMicroResponse queue. - * :param route: route to register with Master + * Called in Consumer (Service Adapter) context. Registers the routing data for self Service + * with the Master router by sending own routing data to Master via CleverMicroResponse queue. + * :param route: route to register with Master """ try: if self.is_open(self.channel) and route.queue: serviceMessage: ServiceMessage = self.service_message_factory.of( ServiceMessageType.ROUTING_DATA, route.as_string(), ANY_RECIPIENT ) - await self.publish_service_message(serviceMessage, self.cm_response_exchange) + await self.publish_service_message( + serviceMessage, self.cm_response_exchange + ) self.add_own_route(route) logging.info(" [DBG] RouterConsumer registered Route: %s", route) else: - logging.warning(" [W] RouterConsumer couldn't register route data: %s channel isn't open.", - CM_RESPONSE_EXCHANGE) + logging.warning( + " [W] RouterConsumer couldn't register route data: %s channel isn't open.", + CM_RESPONSE_EXCHANGE, + ) except Exception as ioe: logging_error("RouterConsumer failed register route data: %s", ioe) async def remove_route_on_cleanup(self, route: AMQRoute) -> bool: """ - * Unregister / remove route from the list - * :param route: route to remove - * :return result of remove operation + * Unregister / remove route from the list + * :param route: route to remove + * :return result of remove operation """ logging.info(" [DBG] RouterConsumer.removeRoute, route: %s", route.as_string()) try: service_message: ServiceMessage = self.service_message_factory.of( - ServiceMessageType.ROUTING_DATA_REMOVE, - route.as_string(), - ANY_RECIPIENT + ServiceMessageType.ROUTING_DATA_REMOVE, route.as_string(), ANY_RECIPIENT ) - if not await self.publish_service_message(service_message, self.cm_response_exchange): - logging.warning(" [W] RouterConsumer couldn't remove current route: %s, channel not open.", - route) + if not await self.publish_service_message( + service_message, self.cm_response_exchange + ): + logging.warning( + " [W] RouterConsumer couldn't remove current route: %s, channel not open.", + route, + ) except Exception as ioe: logging_error("RouterConsumer failed remove current route: %s", ioe) @@ -128,10 +191,19 @@ class RouterConsumer(RouterProducer): def cleanup(self): """ - * Cleanup - de-register routes with master. + * Cleanup - de-register routes with master. """ routes: Set[AMQRoute] = self.get_routes() - logging.info(" [DBG] RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes()) - not_removed = sum(1 for r in (self.remove_route_on_cleanup(route) for route in routes) if not r) + logging.info( + " [DBG] RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes() + ) + not_removed = sum( + 1 + for r in (self.remove_route_on_cleanup(route) for route in routes) + if not r + ) if not_removed > 0: - logging.info(" [W] %d route(s) not removed. self may lead to failed delivery on self route", not_removed) + logging.info( + " [W] %d route(s) not removed. self may lead to failed delivery on self route", + not_removed, + ) diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 32415fa..7172748 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -1,16 +1,22 @@ """ - * Maps the request to RabbitMQ exchange. Handles the service discovery logic. - * - * Router can be either in 'Master' or 'Service (Consumer)' modes. 'Master' mode is the outward - * facing side that receives the requests and routes it to queue associated with requested service. - * 'Service' mode is consumer side for given service, which receives the message from AMQ and - * forwards it to the application. - """ +* Maps the request to RabbitMQ exchange. Handles the service discovery logic. +* +* Router can be either in 'Master' or 'Service (Consumer)' modes. 'Master' mode is the outward +* facing side that receives the requests and routes it to queue associated with requested service. +* 'Service' mode is consumer side for given service, which receives the message from AMQ and +* forwards it to the application. +""" + import logging from aio_pika import Message -from aio_pika.abc import AbstractRobustChannel, AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, \ - AbstractIncomingMessage +from aio_pika.abc import ( + AbstractRobustChannel, + AbstractRobustQueue, + AbstractRobustConnection, + AbstractRobustExchange, + AbstractIncomingMessage, +) from pika.exchange_type import ExchangeType from amqp.adapter.logging_utils import logging_error @@ -18,40 +24,54 @@ from amqp.adapter.service_message_factory import ServiceMessageFactory 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, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, CM_C_AMQ_QUEUE, \ - CM_P_RESP_QUEUE, CM_P_REPLY_TO, ANY_RECIPIENT +from amqp.router.router_base import ( + RouterBase, + CM_REQUEST_EXCHANGE, + CM_RESPONSE_EXCHANGE, + CM_C_AMQ_QUEUE, + CM_P_RESP_QUEUE, + CM_P_REPLY_TO, + ANY_RECIPIENT, +) class RouterProducer(RouterBase): def __init__(self, configuration: AMQConfiguration): """ - * Initialize router - need to supply the config values and RabbitMQ channel. - * :param configuration: adapter's configuration reference. - """ + * Initialize router - need to supply the config values and RabbitMQ channel. + * :param configuration: adapter's configuration reference. + """ super().__init__() self.connection: AbstractRobustConnection | None = None - self.service_message_factory = ServiceMessageFactory(configuration.amq_adapter.service_name) + self.service_message_factory = ServiceMessageFactory( + configuration.amq_adapter.service_name + ) self.amq_configuration: AMQConfiguration = configuration self.channel: AbstractRobustChannel | None = None self.cm_request_exchange: AbstractRobustExchange | None = None self.cm_response_exchange: AbstractRobustExchange | None = None - logging.info(f" [*] RouterProducer.: %s, route: %s", - self.amq_configuration.amq_adapter.service_name, - self.amq_configuration.amq_adapter.route_mapping) + logging.info( + f" [*] RouterProducer.: %s, route: %s", + self.amq_configuration.amq_adapter.service_name, + self.amq_configuration.amq_adapter.route_mapping, + ) async def init_internal_routing_paths(self, channel: AbstractRobustChannel | None): """ - * Initialize internal routes for ServiceMessages. - * The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues - * created to enable the service discovery. - * The Producer mode sends out routing data request on CleverMicroRequest and listens - * on CleverMicroResponse for any service responding with its routing data. - """ + * Initialize internal routes for ServiceMessages. + * The logic here sets a producer and listener on CleverMicroRequest/CleverMicroResponse queues + * created to enable the service discovery. + * The Producer mode sends out routing data request on CleverMicroRequest and listens + * on CleverMicroResponse for any service responding with its routing data. + """ self.channel = channel - logging.info(" [*] RouterProducer.initRoutingPaths, channel open: %s", self.is_open(self.channel)) + logging.info( + " [*] RouterProducer.initRoutingPaths, channel open: %s", + self.is_open(self.channel), + ) if self.is_open(self.channel): try: # **** set up the service discovery internal exchanges **** @@ -68,42 +88,68 @@ class RouterProducer(RouterBase): # 2a. declare a Response queue for Routing Data replies from Services _cm_response_queue: str = self.get_response_queue_name() _response_queue: AbstractRobustQueue = await self.channel.declare_queue( - name=_cm_response_queue, durable=True, exclusive=False, auto_delete=False, - arguments={} + name=_cm_response_queue, + durable=True, + exclusive=False, + auto_delete=False, + arguments={}, ) # 2b. bind queue to CM_RESPONSE_EXCHANGE w/o any conditions (empty routing key) - await _response_queue.bind(exchange=CM_RESPONSE_EXCHANGE, routing_key="#") + await _response_queue.bind( + exchange=CM_RESPONSE_EXCHANGE, routing_key="#" + ) # 2c. listen for incoming RoutingData messages - _c_tag = await _response_queue.consume(callback=self.handle_routing_data_message, no_ack=False) - logging.info(" [%s] Routing master listens for Route updates on %s", _c_tag, _cm_response_queue) + _c_tag = await _response_queue.consume( + callback=self.handle_routing_data_message, no_ack=False + ) + logging.info( + " [%s] Routing master listens for Route updates on %s", + _c_tag, + _cm_response_queue, + ) except Exception as ioe: logging_error("RouterProducer FATAL: cannot initialize, %s", ioe) else: - logging_error("RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open.") + logging_error( + "RouterProducer FATAL: cannot initialize, RabbitMQ channel is not open." + ) async def handle_routing_data_message(self, message: AbstractIncomingMessage): """ - * Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of - * routing data, invokes addRoute or removeRoute depending on the type of the message. - """ - logging.info(f" [x] Received {message.body}, m={message.message_id}, p={message.properties}") + * Handler for CLEVER_MICRO_RESPONSE queue messages. Used in Master mode to handle receipt of + * routing data, invokes addRoute or removeRoute depending on the type of the message. + """ + logging.info( + f" [x] Received {message.body}, m={message.message_id}, p={message.properties}" + ) """ logging.info(" [DBG] ******[%s] ROUTING_DATA %s >>> %s", delivery.getEnvelope().getDeliveryTag(), delivery.getEnvelope().toString(), str(delivery.getBody())) """ - logging.info(" [%s] ACK message now so that it is not being redelivered", message.delivery_tag) + logging.info( + " [%s] ACK message now so that it is not being redelivered", + message.delivery_tag, + ) await message.ack(multiple=False) - cm_message: ServiceMessage = self.service_message_factory.from_bytes(message.body) + cm_message: ServiceMessage = self.service_message_factory.from_bytes( + message.body + ) # ignore the message if it comes from ourselves (even if from different instance) - logging.info(" [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s", - message.delivery_tag, - self.service_message_factory.to_string(cm_message), - self.amq_configuration.amq_adapter.route_mapping, - self.amq_configuration.amq_adapter.service_name, - cm_message.reply_to) - if cm_message.is_valid() and not self.amq_configuration.amq_adapter.service_name == cm_message.reply_to: + logging.info( + " [DBG] ******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s", + message.delivery_tag, + self.service_message_factory.to_string(cm_message), + self.amq_configuration.amq_adapter.route_mapping, + self.amq_configuration.amq_adapter.service_name, + cm_message.reply_to, + ) + if ( + cm_message.is_valid() + and not self.amq_configuration.amq_adapter.service_name + == cm_message.reply_to + ): if cm_message.message_type == ServiceMessageType.ROUTING_DATA: self.add_routes(cm_message.message) if cm_message.message_type == ServiceMessageType.ROUTING_DATA_REMOVE: @@ -111,72 +157,83 @@ class RouterProducer(RouterBase): async def request_routes_from_adapters(self): """ - * Send a broadcast message prompting all listening adapters to reply with own routing data. - """ - logging.info(" [*] RouterProducer.requestRoutesFromAdapters, queue name: %s", CM_REQUEST_EXCHANGE) + * Send a broadcast message prompting all listening adapters to reply with own routing data. + """ + logging.info( + " [*] RouterProducer.requestRoutesFromAdapters, queue name: %s", + CM_REQUEST_EXCHANGE, + ) try: serviceMessage: ServiceMessage = self.service_message_factory.of( ServiceMessageType.ROUTING_DATA_REQUEST, "", ANY_RECIPIENT ) - if not await self.publish_service_message(serviceMessage, self.cm_request_exchange): - logging.warning(" [E] RouterProducer could not request route data: %s channel is not open.", - CM_REQUEST_EXCHANGE) + if not await self.publish_service_message( + serviceMessage, self.cm_request_exchange + ): + logging.warning( + " [E] RouterProducer could not request route data: %s channel is not open.", + CM_REQUEST_EXCHANGE, + ) except Exception as ioe: logging_error("RouterProducer failed request routing data: %s", ioe) - async def publish_service_message(self, - message: ServiceMessage, - exchange: AbstractRobustExchange) -> bool: + async def publish_service_message( + self, message: ServiceMessage, exchange: AbstractRobustExchange + ) -> bool: + """ + * Publishes a service message to the service queue and logs success log entry. + * + * :param message: Service Message to be sent + * :param exchange: target exchange name where to publish the message + * :return True if channel is open, False otherwise + * @throws IOException if something else goes wrong """ - * Publishes a service message to the service queue and logs success log entry. - * - * :param message: Service Message to be sent - * :param exchange: target exchange name where to publish the message - * :return True if channel is open, False otherwise - * @throws IOException if something else goes wrong - """ if self.is_open(self.channel): binary_content: bytes = self.service_message_factory.serialize(message) pika_message: Message = Message( body=binary_content, - content_encoding='utf-8', + content_encoding="utf-8", delivery_mode=2, content_type="application/octet-stream", headers=None, priority=0, - correlation_id=None + correlation_id=None, ) await exchange.publish(message=pika_message, routing_key="") - logging.info(" [x] Service Message Sent to %s, msg: %s", exchange.name, str(binary_content)) + logging.info( + " [x] Service Message Sent to %s, msg: %s", + exchange.name, + str(binary_content), + ) return True return False def get_reply_to_queue_name(self) -> str: """ - * self should create a unique name for self process and also when there's several instances. - * @return unique reply-to queue name. - """ + * self should create a unique name for self process and also when there's several instances. + * @return unique reply-to queue name. + """ return self.amq_configuration.amq_adapter.adapter_prefix() + CM_P_REPLY_TO def get_response_queue_name(self) -> str: """ - * self should create a unique name for self process and also when there's several instances. - * @return unique reply-to queue name. - """ + * self should create a unique name for self process and also when there's several instances. + * @return unique reply-to queue name. + """ return self.amq_configuration.amq_adapter.adapter_prefix() + CM_P_RESP_QUEUE def get_consume_queue_name(self) -> str: """ - * self should create a unique name for self process and also when there's several instances. - * @return unique reply-to queue name. - """ + * self should create a unique name for self process and also when there's several instances. + * @return unique reply-to queue name. + """ return self.amq_configuration.amq_adapter.adapter_prefix() + CM_C_AMQ_QUEUE def is_open(self, _channel: AbstractRobustChannel) -> bool: """ - * Test if the channel is usable (opened). - * :param _channel: the channel - * :return True if the channel is opened. - """ + * Test if the channel is usable (opened). + * :param _channel: the channel + * :return True if the channel is opened. + """ return _channel is not None and not _channel.is_closed diff --git a/amqp/router/test_route_database.py b/amqp/router/test_route_database.py index e9e1df9..863fe75 100644 --- a/amqp/router/test_route_database.py +++ b/amqp/router/test_route_database.py @@ -28,7 +28,9 @@ class TestRouteDatabase(TestCase): def test_get_routes(self): _amq_factory = AMQRouteFactory() - _route: AMQRoute = _amq_factory.from_string("amq-clever-pybook::clever-pybook:clever.pybook.localhost.#:5:0") + _route: AMQRoute = _amq_factory.from_string( + "amq-clever-pybook::clever-pybook:clever.pybook.localhost.#:5:0" + ) _routes = self._database.get_routes() self.assertIsNotNone(_routes) self.assertEqual(3, len(_routes)) @@ -76,8 +78,7 @@ class TestRouteDatabase(TestCase): def test_find_route(self): message = DataMessageFactory.get_instance(111).create_request_message( - "POST", "clever3-book.localhost", - "/aa/bb?cc=d", {}, "" + "POST", "clever3-book.localhost", "/aa/bb?cc=d", {}, "" ) route = self._database.find_route(message.domain, message.path) self.assertIsNotNone(route) diff --git a/amqp/router/test_router_base.py b/amqp/router/test_router_base.py index 51a9bf7..db99df6 100644 --- a/amqp/router/test_router_base.py +++ b/amqp/router/test_router_base.py @@ -44,7 +44,9 @@ class TestRouterBase(TestCase): self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa///bb////cc")) def test_get_routing_key(self): - message = self.factory.create_request_message("POST", "test.me.localhost", "/aa/bb?cc=d", {}, "") + message = self.factory.create_request_message( + "POST", "test.me.localhost", "/aa/bb?cc=d", {}, "" + ) rk = self.base.get_routing_key(message) self.assertEqual("test.me.localhost.aa.bb", rk) @@ -80,8 +82,7 @@ class TestRouterBase(TestCase): def test_find_route_by_message(self): message = DataMessageFactory.get_instance(111).create_request_message( - "POST", "clever3-book.localhost", - "/aa/bb?cc=d", {}, "" + "POST", "clever3-book.localhost", "/aa/bb?cc=d", {}, "" ) _route: AMQRoute = self.base.find_route_by_message(message) self.assertIsNotNone(_route) @@ -91,7 +92,7 @@ class TestRouterBase(TestCase): init_routes: Set[AMQRoute] = self.base.get_routes().copy() self.base.add_routes(ROUTE_6 + ",amq-dlq:DLX:DeadLetterQueue:#:0:0,:::::,:::") modified = self.base.get_routes() - self.assertEqual(len(init_routes)+1, len(modified)) + self.assertEqual(len(init_routes) + 1, len(modified)) wr = self.base.wrap_routes() self.assertFalse("DLX" in wr) diff --git a/amqp/router/utils.py b/amqp/router/utils.py index fc15654..84693f6 100644 --- a/amqp/router/utils.py +++ b/amqp/router/utils.py @@ -5,15 +5,15 @@ pattern = re.compile(r"[^a-zA-Z0-9.#/\-]") def sanitize_as_rabbitmq_name(input_str: str) -> str: matcher = pattern.search(input_str) - token = input_str[:matcher.start()] if matcher else input_str + token = input_str[: matcher.start()] if matcher else input_str return sanitize_as_rabbitmq_domain_name(token) def sanitize_as_rabbitmq_domain_name(token: str) -> str: # Step 1: Replace non-alphanumeric characters (except periods) with a period - step1 = re.sub(r'[^A-Za-z0-9.#]', '.', token) + step1 = re.sub(r"[^A-Za-z0-9.#]", ".", token) # Step 2: Replace multiple consecutive periods with a single period - step2 = re.sub(r'\.+', '.', step1) + step2 = re.sub(r"\.+", ".", step1) # Step 3: Remove trailing '.' character, if applicable step_last_char = len(step2) - 1 - return step2[:-1] if step_last_char > 0 and step2[step_last_char] == '.' else step2 + return step2[:-1] if step_last_char > 0 and step2[step_last_char] == "." else step2 diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 1ddc11f..4b9cb9a 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -23,7 +23,13 @@ from amqp.adapter.file_downloader import on_message from amqp.adapter.logging_utils import logging_error from amqp.adapter.service_message_factory import ServiceMessageFactory from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import DataResponse, DataMessage, MultipartDataMessage, ScalingRequest, ScalingRequestAlert +from amqp.model.model import ( + DataResponse, + DataMessage, + MultipartDataMessage, + ScalingRequest, + ScalingRequestAlert, +) from amqp.model.service_message_type import ServiceMessageType from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient from amqp.router.router_base import ANY_RECIPIENT @@ -31,8 +37,10 @@ from amqp.adapter.serializer import map_as_string # Track the number of messages currently being processed current_parallel_executions = 0 -last_data_message_time = 0 # helps detect IDLE condition -last_backpressure_event_time = 0 # helps prevent flooding the system with backpressure events +last_data_message_time = 0 # helps detect IDLE condition +last_backpressure_event_time = ( + 0 # helps prevent flooding the system with backpressure events +) last_backpressure_event = ScalingRequestAlert.IDLE parallel_workers = int(os.getenv("PARALLEL_WORKERS", default=0)) @@ -45,8 +53,7 @@ executor = concurrent.futures.ThreadPoolExecutor(max_workers=parallel_workers + def collect_trace_info(): trace_map = {} TraceContextTextMapPropagator().inject( - carrier=trace_map, - context=trace.context_api.get_current() + carrier=trace_map, context=trace.context_api.get_current() ) return trace_map @@ -60,15 +67,17 @@ def set_text(carrier, key, value): class DataMessageHandler: - def __init__(self, - service_adapter: CleverThisServiceAdapter, - tracer: Tracer, - message_factory: DataMessageFactory, - service_message_factory: ServiceMessageFactory, - reply_to_exchange: AbstractRobustExchange, - rabbit_mq_client: RabbitMQClient, - amq_configuration: AMQConfiguration, - loop): + def __init__( + self, + service_adapter: CleverThisServiceAdapter, + tracer: Tracer, + message_factory: DataMessageFactory, + service_message_factory: ServiceMessageFactory, + reply_to_exchange: AbstractRobustExchange, + rabbit_mq_client: RabbitMQClient, + amq_configuration: AMQConfiguration, + loop, + ): self.service_adapter: CleverThisServiceAdapter = service_adapter self.tracer = tracer self.message_factory: DataMessageFactory = message_factory @@ -82,7 +91,9 @@ class DataMessageHandler: # Dictionary to store outstanding Futures self.outstanding: Dict[str, asyncio.Future] = {} self.reply_to: str | None = None - self.backpressure_monitor_window = amq_configuration.backpressure.time_window / 1000 # convert to seconds + self.backpressure_monitor_window = ( + amq_configuration.backpressure.time_window / 1000 + ) # convert to seconds global parallel_workers _backpressure_treshold = amq_configuration.backpressure.threshold @@ -104,7 +115,10 @@ class DataMessageHandler: now = time.time() event: ScalingRequestAlert = ScalingRequestAlert.UPDATE if now - last_backpressure_event_time >= self.backpressure_monitor_window: - if current_parallel_executions == 0 and now - last_data_message_time > self.backpressure_monitor_window: + if ( + current_parallel_executions == 0 + and now - last_data_message_time > self.backpressure_monitor_window + ): event = ScalingRequestAlert.IDLE elif current_parallel_executions > parallel_workers - 2: event = ScalingRequestAlert.OVERLOAD @@ -112,29 +126,36 @@ class DataMessageHandler: last_backpressure_event_time = now last_backpressure_event = event last_backpressure_event_time = now - logging.info(f"Backpressure event: {event}, threads: {current_parallel_executions} of {parallel_workers}") + logging.info( + f"Backpressure event: {event}, threads: {current_parallel_executions} of {parallel_workers}" + ) scaling_request: ScalingRequest = ScalingRequest( - swarm_service_id, swarm_task_id, - parallel_workers, parallel_workers - current_parallel_executions, - event + swarm_service_id, + swarm_task_id, + parallel_workers, + parallel_workers - current_parallel_executions, + event, ) # Address the message to any adapter capable of supporting BACKPRESSURE request scale_up_request = self.service_message_factory.of( ServiceMessageType.BACKPRESSURE, scaling_request.to_json(), - self.reply_to if self.reply_to else ANY_RECIPIENT + self.reply_to if self.reply_to else ANY_RECIPIENT, ) asyncio.run_coroutine_threadsafe( self.rabbit_mq_client.router.publish_service_message( - scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange + scale_up_request, + exchange=self.rabbit_mq_client.router.cm_response_exchange, ), - loop=self.loop + loop=self.loop, ) await asyncio.sleep(self.backpressure_monitor_window) # sleep in seconds async def inbound_data_message_callback(self, message: AbstractIncomingMessage): - threading.Thread(target=lambda: asyncio.run(self.run_callback_thread(message))).start() + threading.Thread( + target=lambda: asyncio.run(self.run_callback_thread(message)) + ).start() async def run_callback_thread(self, message: AbstractIncomingMessage): """ @@ -151,12 +172,16 @@ class DataMessageHandler: if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST: await self.run_http_request_magic(message, amq_message, self.loop) else: - logging_error("Unknown MAGIC value: [%s], don't know how to interpret the message", - amq_message.magic) + logging_error( + "Unknown MAGIC value: [%s], don't know how to interpret the message", + amq_message.magic, + ) asyncio.run_coroutine_threadsafe(message.nack(requeue=False), self.loop) return else: - logging_error("######[%s] No valid AMQ Message present.", message.delivery_tag) + logging_error( + "######[%s] No valid AMQ Message present.", message.delivery_tag + ) self.record_duration(start, message.delivery_tag) asyncio.run_coroutine_threadsafe(message.ack(multiple=False), self.loop) @@ -164,30 +189,35 @@ class DataMessageHandler: logging.warning("Warning: Capacity close to depleted!") # Send a scaling request to the Management service scaling_request: ScalingRequest = ScalingRequest( - swarm_service_id, swarm_task_id, - parallel_workers, parallel_workers - current_parallel_executions, - ScalingRequestAlert.OVERLOAD + swarm_service_id, + swarm_task_id, + parallel_workers, + parallel_workers - current_parallel_executions, + ScalingRequestAlert.OVERLOAD, ) # Address the message to any adapter capable of supporting BACKPRESSURE request scale_up_request = self.service_message_factory.of( ServiceMessageType.BACKPRESSURE, scaling_request.to_json(), - self.reply_to if self.reply_to else ANY_RECIPIENT + self.reply_to if self.reply_to else ANY_RECIPIENT, ) asyncio.run_coroutine_threadsafe( self.rabbit_mq_client.router.publish_service_message( - scale_up_request, exchange=self.rabbit_mq_client.router.cm_response_exchange + scale_up_request, + exchange=self.rabbit_mq_client.router.cm_response_exchange, ), - loop=_loop + loop=_loop, ) # update / reset time-window so that the OVERLOAD is not sent too often global last_data_message_time last_datadeliveryTag_message_time = time.time() - async def run_http_request_magic(self, - message: AbstractIncomingMessage, - amq_message: DataMessage, - _loop: AbstractEventLoop) -> None: + async def run_http_request_magic( + self, + message: AbstractIncomingMessage, + amq_message: DataMessage, + _loop: AbstractEventLoop, + ) -> None: """ Synchronous version of on_message method. This is a wrapper around the async on_message method. It also counts parallel executions and handles backpressure. @@ -200,7 +230,9 @@ class DataMessageHandler: # Increment the counter for parallel executions current_parallel_executions += 1 - logging.info(f"PARALLEL: Current Capacity [{current_parallel_executions}] of [{parallel_workers}]") + logging.info( + f"PARALLEL: Current Capacity [{current_parallel_executions}] of [{parallel_workers}]" + ) if message.reply_to: self.reply_to = message.reply_to if current_parallel_executions > parallel_workers - 2: @@ -211,20 +243,31 @@ class DataMessageHandler: response: DataResponse = await self.service_adapter.on_message(a_message) print(f"Result in thread: {response}") try: - logging.info(" [*] ######[%s] AMQ Message Reply.1, To=%s, msgId=%s", - message.delivery_tag, message.reply_to, response.id) + logging.info( + " [*] ######[%s] AMQ Message Reply.1, To=%s, msgId=%s", + message.delivery_tag, + message.reply_to, + response.id, + ) await self.send_reply(message.reply_to, response) - logging.info(" [*] ######[%s] AMQ Message Reply.2, To=%s", - message.delivery_tag, message.reply_to) + logging.info( + " [*] ######[%s] AMQ Message Reply.2, To=%s", + message.delivery_tag, + message.reply_to, + ) except Exception as e: logging_error(f"Processing message: {e}") - asyncio.run_coroutine_threadsafe(message.nack(requeue=False), loop=_loop) + asyncio.run_coroutine_threadsafe( + message.nack(requeue=False), loop=_loop + ) if isinstance(amq_message, MultipartDataMessage): # remove the downloaded files from the output_dir for file_id, filename in amq_message.extra_data.items(): try: - logging.info(f" [*][{message.delivery_tag}] Deleting file: {file_id} in {filename}") + logging.info( + f" [*][{message.delivery_tag}] Deleting file: {file_id} in {filename}" + ) os.remove(filename) except OSError as e: logging_error(f"Deleting file {filename}: {e}") @@ -234,34 +277,50 @@ class DataMessageHandler: logging_error(f"Request handler: {e}") return - - async def reconstitute_data_message(self, message: AbstractIncomingMessage) -> DataMessage | None: + async def reconstitute_data_message( + self, message: AbstractIncomingMessage + ) -> DataMessage | None: amq_message: DataMessage | None = None delivery_tag = message.delivery_tag - addressing: int = message.headers.get('clevermicro.addressing', 0) + addressing: int = message.headers.get("clevermicro.addressing", 0) if addressing == 0: amq_message = DataMessageFactory.from_bytes(message.body) else: - amq_message = await DataMessageFactory.from_stream(message.body, rabbitmq_url=self.rabbit_mq_url, loop=self.loop) - logging.info(f" [DBG] ######[{delivery_tag}] AMQ Message from stream: {amq_message}") + amq_message = await DataMessageFactory.from_stream( + message.body, rabbitmq_url=self.rabbit_mq_url, loop=self.loop + ) + logging.info( + f" [DBG] ######[{delivery_tag}] AMQ Message from stream: {amq_message}" + ) if amq_message is not None: extra_data = await on_message( message=message, json_data=amq_message.body(), rabbitmq_url=self.rabbit_mq_url, loop=self.loop, - output_dir=self.output_dir) - logging.info(f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}") + output_dir=self.output_dir, + ) + logging.info( + f" [x] ######[{delivery_tag}] Multipart download result: {extra_data}" + ) amq_message = MultipartDataMessage(amq_message, extra_data) if amq_message: - logging.info(" [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, %s:%s, traceInfo=%s", - message.delivery_tag, message.consumer_tag, amq_message.id, - amq_message.method, amq_message.path, map_as_string(amq_message.trace_info)) + logging.info( + " [*] ######[%s] AMQ Message received, ctag=%s, msgId=%s, %s:%s, traceInfo=%s", + message.delivery_tag, + message.consumer_tag, + amq_message.id, + amq_message.method, + amq_message.path, + map_as_string(amq_message.trace_info), + ) return amq_message def ensure_trace_info(self, amq_message: DataMessage): propagator = TraceContextTextMapPropagator() - context = propagator.extract(carrier=amq_message.trace_info, context=trace.context_api.get_current()) + context = propagator.extract( + carrier=amq_message.trace_info, context=trace.context_api.get_current() + ) self.tracer.start_span(name="AMQ-Adapter-Consumer", context=context) logging.info(" [*] CTX=%s", context) # current_span = self.tracer.start_span("AMQ-Adapter.Processing.Request", parent=parent_span) @@ -274,7 +333,11 @@ class DataMessageHandler: def record_duration(self, start, delivery_tag): global last_data_message_time last_data_message_time = time.time() - logging.info(" [x] ######[%s] Done, single ACK: %sms.", delivery_tag, last_data_message_time - start) + logging.info( + " [x] ######[%s] Done, single ACK: %sms.", + delivery_tag, + last_data_message_time - start, + ) async def send_reply(self, reply_to: str, response: DataResponse): """ @@ -288,15 +351,27 @@ class DataMessageHandler: pika_message: Message = Message( body=DataResponseFactory.serialize(response), correlation_id=response.id, - content_type=response.content_type[0] - if isinstance(response.content_type, list) else response.content_type + content_type=( + response.content_type[0] + if isinstance(response.content_type, list) + else response.content_type + ), + ) + logging.info( + " [*] ###### Sending Reply.1 To=%s, msgId=%s", reply_to, response.id + ) + _res = asyncio.run_coroutine_threadsafe( + self.reply_to_exchange.publish( + message=pika_message, routing_key=reply_to + ), + self.loop, + ) + logging.info( + " [*] ###### DONE EXCH PUB(%s) Sending Reply.2 To=%s, res=%s", + self.reply_to_exchange.name, + reply_to, + _res, ) - logging.info(" [*] ###### Sending Reply.1 To=%s, msgId=%s", reply_to, response.id) - _res = asyncio.run_coroutine_threadsafe(self.reply_to_exchange.publish( - message=pika_message, - routing_key=reply_to - ), self.loop) - logging.info(" [*] ###### DONE EXCH PUB(%s) Sending Reply.2 To=%s, res=%s", self.reply_to_exchange.name, reply_to, _res) async def reply_received_callback(self, message: AbstractIncomingMessage): """ @@ -312,7 +387,9 @@ class DataMessageHandler: logging.info(debug) # TODO - figure out implementation of this part, as this is sending reply back either to a service, # or to the user - logging.info(f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}") + logging.info( + f"OUTSTANDING RECV: {self.outstanding}, size={len(self.outstanding)}" + ) future: Future = self.outstanding.pop(reply_id, None) # if reply is None: diff --git a/amqp/service/amq_service.py b/amqp/service/amq_service.py index 0ff04fc..9ff8260 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -19,14 +19,15 @@ from amqp.service.tracing import initialize_trace logging.basicConfig(level=logging.DEBUG) logging.info(f"CWD = {os.getcwd()}") -logging_conf_path = os.path.join(os.getcwd(), 'amqp', 'service', 'logging.conf') +logging_conf_path = os.path.join(os.getcwd(), "amqp", "service", "logging.conf") if os.path.exists(logging_conf_path): logging.config.fileConfig(logging_conf_path) else: - logging_conf_path = os.path.join(os.getcwd(), 'logging.conf') + logging_conf_path = os.path.join(os.getcwd(), "logging.conf") if os.path.exists(logging_conf_path): logging.config.fileConfig(logging_conf_path) + class AMQService: CLEVER_AMQP_CLIENT_VERSION = os.getenv("CLEVER_AMQP_CLIENT_VERSION") MAX_WAIT = 300 # defaults to 5 minutes (300 seconds) @@ -38,7 +39,9 @@ class AMQService: self.loop = asyncio.get_event_loop() self.amq_configuration = amq_configuration - self.data_message_factory = DataMessageFactory.get_instance(self.amq_configuration.amq_adapter.generator_id) + self.data_message_factory = DataMessageFactory.get_instance( + self.amq_configuration.amq_adapter.generator_id + ) self.rabbit_mq_client = RabbitMQClient(self.amq_configuration) self.tracer = initialize_trace(name=amq_configuration.amq_adapter.service_name) self.amq_data_message_handler = None @@ -54,7 +57,9 @@ class AMQService: self.loop.run_forever() async def init(self, loop, service_adapter): - _reply_to_exchange: AbstractRobustExchange = await self.rabbit_mq_client.init(loop) + _reply_to_exchange: AbstractRobustExchange = await self.rabbit_mq_client.init( + loop + ) self.amq_data_message_handler = DataMessageHandler( service_adapter, self.tracer, @@ -63,7 +68,7 @@ class AMQService: _reply_to_exchange, self.rabbit_mq_client, self.amq_configuration, - loop=loop + loop=loop, ) await self.register_routes() await self.rabbit_mq_client.request_routes() @@ -85,7 +90,8 @@ class AMQService: if AMQRouteFactory.validate(route_mapping): try: await self.rabbit_mq_client.register_inbound_routes( - route_mapping, self.amq_data_message_handler.inbound_data_message_callback + route_mapping, + self.amq_data_message_handler.inbound_data_message_callback, ) return await self.rabbit_mq_client.register_data_reply_callback( self.amq_data_message_handler.reply_received_callback @@ -93,7 +99,10 @@ class AMQService: except Exception as e: raise RuntimeError(e) elif route_mapping: - logging.info(" [W] route [%s] failed validation, no routes were registered.", route_mapping) + logging.info( + " [W] route [%s] failed validation, no routes were registered.", + route_mapping, + ) return None def send_message(self, message: DataMessage, future: Future): @@ -101,11 +110,3 @@ class AMQService: def remove_outstanding_future(self, message_id: str): self.amq_data_message_handler.outstanding.pop(message_id, None) - - -# if __name__ == "__main__": - # amq_configuration: AMQConfiguration = AMQConfiguration('../config/application.properties.local') - # amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAmqpAdapter(amq_configuration)) - # amq_service.rabbit_mq_client.channel.start_consuming() - # amq_service.consumer_thread.cancel() - # amq_service.rabbit_mq_client.connection.close() diff --git a/amqp/service/delete_queues.py b/amqp/service/delete_queues.py index a33b57d..52c9faf 100644 --- a/amqp/service/delete_queues.py +++ b/amqp/service/delete_queues.py @@ -1,7 +1,13 @@ import requests -def delete_queues_with_prefix_http(host='localhost', port=15672, username='springuser', password='TheCleverWho', prefix='cm-'): +def delete_queues_with_prefix_http( + host="localhost", + port=15672, + username="springuser", + password="TheCleverWho", + prefix="cm-", +): """ CLI utility. Delete queues using RabbitMQ HTTP API. """ @@ -11,7 +17,7 @@ def delete_queues_with_prefix_http(host='localhost', port=15672, username='sprin response = requests.get(base_url, auth=(username, password)) response.raise_for_status() - queues = [q['name'] for q in response.json() if q['name'].startswith(prefix)] + queues = [q["name"] for q in response.json() if q["name"].startswith(prefix)] if not queues: print(f"No queues found with prefix '{prefix}'") @@ -38,5 +44,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_') + delete_queues_with_prefix_http(prefix="amq.gen--") + delete_queues_with_prefix_http(prefix="amq_") diff --git a/amqp/service/tracing.py b/amqp/service/tracing.py index d54a54d..4d4210f 100644 --- a/amqp/service/tracing.py +++ b/amqp/service/tracing.py @@ -11,13 +11,17 @@ from opentelemetry.sdk.resources import SERVICE_NAME, Resource import os -def initialize_trace(app = None, name = None) -> Tracer: +def initialize_trace(app=None, name=None) -> Tracer: # Resource can be required for some backends, e.g. Jaeger or Prometheus # If resource wouldn't be set - traces wouldn't appear in Jaeger. # The 'name' value is the name shown in Jaeger. - resource = Resource(attributes={ - SERVICE_NAME: name if name else os.environ.get('APPLICATION_NAME', 'Python App') - }) + resource = Resource( + attributes={ + SERVICE_NAME: ( + name if name else os.environ.get("APPLICATION_NAME", "Python App") + ) + } + ) # # 1. Tracing (Jaeger) configuration # ---------------------------------- @@ -82,7 +86,6 @@ def initialize_trace(app = None, name = None) -> Tracer: # of that parent span, indicating visually the progression of requests via different services # for as long as the current span is provided in the traceparent header to next service. - # 2. Metrics (Prometheus) configuration # ------------------------------------- # @@ -97,4 +100,3 @@ def initialize_trace(app = None, name = None) -> Tracer: metrics.set_meter_provider(provider) return trace.get_tracer(name) -