From 15da316a1facc6dbc1df08d9bea0d57803af2142 Mon Sep 17 00:00:00 2001 From: Stanislav Hejny Date: Thu, 8 May 2025 23:52:33 +0100 Subject: [PATCH] #43 - fix the issues reported by flake8 --- .flake8 | 4 + .pre-commit-config.yaml | 21 +++++ amqp/adapter/cleverthis_service_adapter.py | 29 ++++--- amqp/adapter/file_handler.py | 16 ++-- amqp/adapter/service_message_factory.py | 3 +- amqp/config/amq_configuration.py | 2 +- amqp/model/model.py | 82 ++++++++----------- amqp/router/route_database.py | 1 - amqp/router/router_base.py | 5 +- amqp/router/router_consumer.py | 10 +-- amqp/router/router_producer.py | 22 +++-- amqp/service/amq_message_handler.py | 18 ++-- .../test_cleverthis_service_adapter.py | 10 +-- 13 files changed, 115 insertions(+), 108 deletions(-) create mode 100644 .flake8 create mode 100644 .pre-commit-config.yaml diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..8ace7dc --- /dev/null +++ b/.flake8 @@ -0,0 +1,4 @@ +[flake8] +max-line-length = 88 +ignore = E501, W503, E203 +exclude = .git,__pycache__,docs/source/conf.py,old,build,dist,.venv diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..3fa6376 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,21 @@ +repos: +- repo: https://github.com/pre-commit/pre-commit-hooks + rev: v4.2.0 + hooks: + - id: check-yaml + - id: end-of-file-fixer + - id: trailing-whitespace +- repo: https://github.com/PyCQA/isort + rev: 6.0.1 + hooks: + - id: isort + args: ["--profile", "black"] +- repo: https://github.com/psf/black + rev: 25.1.0 + hooks: + - id: black + language_version: python3.11 +- repo: https://github.com/PyCQA/flake8 + rev: 7.2.0 + hooks: + - id: flake8 diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 37049ad..657b41b 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -7,26 +7,25 @@ import asyncio import inspect import json from asyncio import AbstractEventLoop -from typing import Dict, List, Callable, Coroutine, Any +from dataclasses import dataclass +from typing import Any, Callable, Coroutine, Dict, List from aio_pika.abc import ( + AbstractExchange, AbstractRobustChannel, AbstractRobustConnection, - AbstractExchange, ) -from aiohttp import ClientSession, ClientResponse -from dataclasses import dataclass +from aiohttp import ClientResponse, ClientSession from fastapi import HTTPException from pydantic import BaseModel from starlette.responses import FileResponse -from amqp.adapter import serializer, pydantic_serializer -from amqp.adapter.data_parser import AMQDataParser +from amqp.adapter import serializer from amqp.adapter.data_response_factory import DataResponseFactory from amqp.adapter.file_handler import FileHandler, StreamingFileHandler -from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info +from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import DataMessage, DataResponse, AMQErrorMessage +from amqp.model.model import AMQErrorMessage, DataMessage, DataResponse from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE from amqp.router.utils import await_result @@ -97,7 +96,7 @@ async def _convert_file_response( _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) + asyncio.run_coroutine_threadsafe(_channel.close(), loop) return json.dumps( { "files": [ @@ -262,7 +261,7 @@ class CleverThisServiceAdapter: :param auth_user: user authenticated by the API Gateway, as UserSchem :return: DataResponse to be sent back to the API Gateway via AMQP """ - return await self.handle_possible_form( + return await self._authorize_forward_request( message, self.session.put( f"http://{self.service_host}:{self.service_port}{message.path}", @@ -280,7 +279,7 @@ class CleverThisServiceAdapter: :param auth_user: user authenticated by the API Gateway, as UserSchem :return: DataResponse to be sent back to the API Gateway via AMQP """ - return await self.handle_possible_form( + return await self._authorize_forward_request( message, self.session.post( f"http://{self.service_host}:{self.service_port}{message.path}", @@ -290,10 +289,14 @@ class CleverThisServiceAdapter: auth_user, ) - async def handle_possible_form( + async def _authorize_forward_request( self, message: DataMessage, request_coroutine, auth_user: Any ) -> DataResponse: - _args = AMQDataParser.parse_request_body(message) + """ + This method is used to authorize the request and forward it to the CleverThis Service API. + However, it is only used for loose / 3rd party mode of coupling with the Service + """ + # AMQDataParser.parse_request_body(message) return await request_coroutine async def head(self, message: DataMessage, auth_user: Any) -> DataResponse: diff --git a/amqp/adapter/file_handler.py b/amqp/adapter/file_handler.py index 0ebf992..a2eacf9 100644 --- a/amqp/adapter/file_handler.py +++ b/amqp/adapter/file_handler.py @@ -4,22 +4,22 @@ implements the file downloader for the amqp adapter, where file is sent in chunk import asyncio import json -import os import logging -from asyncio import Future, AbstractEventLoop +import os +from asyncio import AbstractEventLoop from collections import defaultdict from pathlib import Path -from aio_pika import connect_robust, Message +from aio_pika import Message, connect_robust from aio_pika.abc import ( - AbstractIncomingMessage, - AbstractRobustChannel, - AbstractQueue, AbstractExchange, + AbstractIncomingMessage, + AbstractQueue, + AbstractRobustChannel, DeliveryMode, ) -from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info +from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info from amqp.router.utils import await_result # Global dictionary to track completed downloads @@ -166,7 +166,6 @@ class StreamingFileHandler(FileHandler): message_data (dict): The entire message 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"] @@ -281,7 +280,6 @@ class StreamingFileHandler(FileHandler): loop: The asyncio event loop of the RabbitMQ API output_dir """ - global download_locations try: _message_data = json.loads(message.body.decode()) _amq_message_data = json.loads(json_data.decode()) diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py index 2ea6af6..4d0c797 100644 --- a/amqp/adapter/service_message_factory.py +++ b/amqp/adapter/service_message_factory.py @@ -1,13 +1,12 @@ import base64 -import logging from amqp.adapter.data_message_factory import DataMessageFactory from amqp.adapter.logging_utils import logging_error, logging_info +from amqp.adapter.serializer import map_as_string, parse_map_string from amqp.adapter.trace_info_adapter import TraceInfoAdapter from amqp.model.model import ServiceMessage from amqp.model.service_message_type import ServiceMessageType from amqp.model.snowflake_id import SnowflakeId -from amqp.adapter.serializer import parse_map_string, map_as_string RS = "|" SPLIT_RS = r"\|" diff --git a/amqp/config/amq_configuration.py b/amqp/config/amq_configuration.py index 4c2ba19..905e356 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -5,7 +5,7 @@ import os from amqp.adapter.logging_utils import logging_error, logging_warning """ -Convert configuration from properties file to an object. +Convert configuration from properties file to an object. """ diff --git a/amqp/model/model.py b/amqp/model/model.py index 0598461..9ee4bee 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -1,8 +1,8 @@ import base64 import json -from enum import Enum -from typing import List, Dict from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List from amqp.model.service_message_type import ServiceMessageType from amqp.model.snowflake_id import SnowflakeId @@ -16,9 +16,24 @@ Define the structure of messages used in Clever Micro """ +@dataclass class CleverMicroMessage: DEFAULT_CONTENT_TYPE = ["application/octet-stream"] + """ + The common structure of a message that transfers the user request to the Service and back. All data related messages + conform to this base structure: + + magic - identifies the message type (REST, RPC, DataResponse) + id - unique message ID + m_field - method name (for REST style calls) or service name (for RPC style calls) or response code for DataResponse + d_field - domain name (for REST style calls) or module name (for RPC style calls) or error code for DataResponse + p_field - path (for REST style calls) or method name (for RPC style calls) or error cause for DataResponse + headers - headers of the message (e.g. Content-Type, Accept, etc.) + trace_info - trace information (e.g. trace ID, span ID, etc. for Jaeger trace monitor) + base64body - the body of the message encoded in base64 + """ + def __init__( self, magic: str, @@ -66,11 +81,19 @@ class CleverMicroMessage: def body(self) -> bytes: return base64.b64decode(self.base64body()) - def contentType(self) -> str: + def content_type(self) -> str: return self.headers().get("Content-Type", self.DEFAULT_CONTENT_TYPE)[0] class DataMessage(CleverMicroMessage): + """ + The class represents a structure of a message that transfers the user request to the Service. + It may also be used to make a call/request between services. + The response to this call is in format of DataResponse class. + + See DataMessageFactory for code to instantiate, serialize and deserialize the class + """ + def __init__( self, magic: str, @@ -95,8 +118,17 @@ class DataMessage(CleverMicroMessage): def path(self) -> str: return self.p_field() + def routing_key(self) -> str: + return sanitize_as_rabbitmq_name(f"{self.d_field()}.{self.p_field()}") + class DataResponse(CleverMicroMessage): + """ + The response message to REST & RPC style calls (note: the request is made with DataMessage) + + See DataResponseFactory for code to instantiate, serialize and deserialize the class + """ + def __init__( self, magic: str, @@ -179,31 +211,6 @@ class AMQRoute: return self.__str__() -@dataclass -class DataMessage: - """ - The class represents a structure of a message that transfers the user request to the Service. - It may also be used to make a call/request between services. - 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 - domain: str - path: str - headers: dict[str, List[str]] - trace_info: dict[str, str] - base64_body: bytes - - def routing_key(self) -> str: - return sanitize_as_rabbitmq_name(f"{self.domain}.{self.path}") - - def body(self) -> bytes: - return base64.b64decode(self.base64_body) - - class MultipartDataMessage(DataMessage): def __init__(self, message: DataMessage, extra_data: dict): super().__init__( @@ -219,25 +226,6 @@ class MultipartDataMessage(DataMessage): self.extra_data = extra_data -@dataclass -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 - response: bytes - error: str | None - error_cause: str | None - trace_info: dict[str, str] - - def body(self) -> bytes: - return base64.b64decode(self.response) - - # Ensure backward compatibility AMQResponse = DataResponse diff --git a/amqp/router/route_database.py b/amqp/router/route_database.py index 678ae16..637b793 100644 --- a/amqp/router/route_database.py +++ b/amqp/router/route_database.py @@ -1,4 +1,3 @@ -import logging import re from typing import Set diff --git a/amqp/router/router_base.py b/amqp/router/router_base.py index 82caecd..a683db1 100644 --- a/amqp/router/router_base.py +++ b/amqp/router/router_base.py @@ -1,8 +1,7 @@ -import logging from typing import Callable, List, Set -from amqp.adapter.amq_route_factory import AMQRouteFactory, NULL_ROUTE -from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info +from amqp.adapter.amq_route_factory import NULL_ROUTE, AMQRouteFactory +from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info from amqp.model.model import AMQRoute, DataMessage from amqp.router.route_database import RouteDatabase from amqp.router.utils import sanitize_as_rabbitmq_name diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py index fecfa75..23384b8 100644 --- a/amqp/router/router_consumer.py +++ b/amqp/router/router_consumer.py @@ -1,26 +1,26 @@ -import logging from typing import Set from aio_pika.abc import ( + AbstractIncomingMessage, AbstractRobustChannel, AbstractRobustQueue, - AbstractIncomingMessage, ) + from amqp.adapter.amq_route_factory import AMQRouteFactory from amqp.adapter.logging_utils import ( + logging_debug, logging_error, logging_info, - logging_debug, logging_warning, ) from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import ServiceMessage, AMQRoute +from amqp.model.model import AMQRoute, ServiceMessage from amqp.model.service_message_type import ServiceMessageType from amqp.router.router_base import ( + ANY_RECIPIENT, CM_C_REQ_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, - ANY_RECIPIENT, ) from amqp.router.router_producer import RouterProducer diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 174b502..054a541 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -7,22 +7,20 @@ * forwards it to the application. """ -import logging - from aio_pika import Message from aio_pika.abc import ( + AbstractIncomingMessage, AbstractRobustChannel, - AbstractRobustQueue, AbstractRobustConnection, AbstractRobustExchange, - AbstractIncomingMessage, + AbstractRobustQueue, ) from pika.exchange_type import ExchangeType from amqp.adapter.logging_utils import ( + logging_debug, logging_error, logging_info, - logging_debug, logging_warning, ) from amqp.adapter.service_message_factory import ServiceMessageFactory @@ -30,13 +28,13 @@ 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, + ANY_RECIPIENT, + CM_C_AMQ_QUEUE, + CM_P_REPLY_TO, + CM_P_RESP_QUEUE, CM_REQUEST_EXCHANGE, CM_RESPONSE_EXCHANGE, - CM_C_AMQ_QUEUE, - CM_P_RESP_QUEUE, - CM_P_REPLY_TO, - ANY_RECIPIENT, + RouterBase, ) @@ -58,7 +56,7 @@ class RouterProducer(RouterBase): self.cm_response_exchange: AbstractRobustExchange | None = None logging_info( - f"RouterProducer.: %s, route: %s", + "RouterProducer.: %s, route: %s", self.amq_configuration.amq_adapter.service_name, self.amq_configuration.amq_adapter.route_mapping, ) @@ -126,7 +124,7 @@ class RouterProducer(RouterBase): * routing data, invokes addRoute or removeRoute depending on the type of the message. """ logging_info( - f"[%s] Received ROUTING DATA: [%s], msgId=%s", + "[%s] Received ROUTING DATA: [%s], msgId=%s", message.delivery_tag, message.body, message.message_id, diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index 29aa07a..c099e96 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -1,32 +1,31 @@ import asyncio +import os import threading import time -import os -from asyncio import Future, AbstractEventLoop +from asyncio import AbstractEventLoop, Future from typing import Dict from aio_pika import Message from aio_pika.abc import AbstractIncomingMessage, AbstractRobustExchange - from opentelemetry import trace -from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator from opentelemetry.trace import Tracer, TraceState +from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator from amqp.adapter.backpressure_handler import BackpressureHandler +from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter from amqp.adapter.data_message_factory import DataMessageFactory from amqp.adapter.data_response_factory import DataResponseFactory -from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage -from amqp.adapter.file_handler import StreamingFileHandler, FileHandler -from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info +from amqp.adapter.file_handler import FileHandler, StreamingFileHandler +from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info +from amqp.adapter.serializer import map_as_string from amqp.adapter.service_message_factory import ServiceMessageFactory from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import ( - DataResponse, DataMessage, + DataResponse, MultipartDataMessage, ) from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient -from amqp.adapter.serializer import map_as_string def collect_trace_info(): @@ -266,7 +265,6 @@ class DataMessageHandler: :return: """ reply_id = message.correlation_id - content_type = message.content_type delivery_tag = message.delivery_tag debug = f"######[{delivery_tag}] (RPC REPLY) env: {message.headers}, props: {message.properties}, BODY: {message.body.decode()}" logging_debug(debug) diff --git a/tests/adapter/test_cleverthis_service_adapter.py b/tests/adapter/test_cleverthis_service_adapter.py index 90be99e..f4aafd4 100644 --- a/tests/adapter/test_cleverthis_service_adapter.py +++ b/tests/adapter/test_cleverthis_service_adapter.py @@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch from aiohttp import ClientResponse -from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage +from amqp.adapter.cleverthis_service_adapter import AMQMessage, CleverThisServiceAdapter from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import DataMessage, DataResponse @@ -174,8 +174,8 @@ class TestCleverThisServiceAdapter(unittest.TestCase): with patch.object( self.adapter, - "handle_possible_form", - wraps=self.adapter.handle_possible_form, + "_authorize_forward_request", + wraps=self.adapter._authorize_forward_request, ) as mock_handler: response = await self.adapter.put(mock_message, None) @@ -199,8 +199,8 @@ class TestCleverThisServiceAdapter(unittest.TestCase): with patch.object( self.adapter, - "handle_possible_form", - wraps=self.adapter.handle_possible_form, + "_authorize_forward_request", + wraps=self.adapter._authorize_forward_request, ) as mock_handler: response = await self.adapter.post(mock_message, None)