diff --git a/amqp/adapter/backpressure_handler.py b/amqp/adapter/backpressure_handler.py index 1fca9e2..005f3c0 100644 --- a/amqp/adapter/backpressure_handler.py +++ b/amqp/adapter/backpressure_handler.py @@ -163,10 +163,7 @@ class BackpressureHandler: CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2 IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90 self.average_cpu_usage = RunningAverage( - int( - max(_overload_duration_millis, _idle_duration_millis) - // _monitor_interval - ) + int(max(_overload_duration_millis, _idle_duration_millis) // _monitor_interval) ) async def _backpressure_monitor(): @@ -283,15 +280,11 @@ class BackpressureHandler: if not self.exchange: async def _wrap_rabbit_mq_api_init(channel): - _exchange = await channel.get_exchange( - name="cleverthis.clevermicro.management" - ) + _exchange = await channel.get_exchange(name="cleverthis.clevermicro.management") return _exchange self.exchange = await await_result( - asyncio.run_coroutine_threadsafe( - _wrap_rabbit_mq_api_init(self.channel), self.loop - ) + asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api_init(self.channel), self.loop) ) if self.exchange: @@ -319,6 +312,4 @@ class BackpressureHandler: return True return False - await await_future( - asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop) - ) + await await_future(asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)) diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 657b41b..e7d00da 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -6,8 +6,10 @@ Methods here will invoke the actual CleverThis Service code, which are passed as import asyncio import inspect import json +import pathlib +import re from asyncio import AbstractEventLoop -from dataclasses import dataclass +from collections import defaultdict from typing import Any, Callable, Coroutine, Dict, List from aio_pika.abc import ( @@ -16,44 +18,24 @@ from aio_pika.abc import ( AbstractRobustConnection, ) from aiohttp import ClientResponse, ClientSession -from fastapi import HTTPException +from fastapi import HTTPException, UploadFile +from fastapi.routing import APIRoute +from fastapi.security import OAuth2PasswordRequestForm from pydantic import BaseModel +from starlette.datastructures import Headers from starlette.responses import FileResponse from amqp.adapter import serializer +from amqp.adapter.data_parser import AMQDataParser from amqp.adapter.data_response_factory import DataResponseFactory from amqp.adapter.file_handler import FileHandler, StreamingFileHandler from amqp.adapter.logging_utils import logging_debug, logging_error, logging_info from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import AMQErrorMessage, DataMessage, DataResponse +from amqp.model.model import AMQErrorMessage, AMQMessage, DataMessage, DataResponse from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE from amqp.router.utils import await_result -@dataclass -class AMQMessage(DataMessage): - """ - Expand the core DataMessage type to include the channel, which is required to create a dedicated queue - in case the response should initiate file download. - """ - - def __init__(self, data_message: DataMessage, connection: AbstractRobustConnection): - # Copy all fields from DataMessage - self.magic = data_message.magic - self.id = data_message.id - self.method = data_message.method - self.domain = data_message.domain - self.path = data_message.path - self.headers = data_message.headers - self.trace_info = data_message.trace_info - 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 {} - ) - - async def _convert_file_response( obj: FileResponse, connection: AbstractRobustConnection, @@ -62,15 +44,16 @@ async def _convert_file_response( ) -> 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. + 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, ) -> tuple[AbstractRobustChannel, AbstractExchange, str]: - # Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with correct event loop - # This is needed because the RabbitMQ API is not thread-safe and needs to be called in the context of - # the event loop the first RabbitMQ connection was created with. + # 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 event loop the first RabbitMQ connection was created with. _channel = await connection.channel() _exchange_name: str = AMQ_REPLY_TO_EXCHANGE _exchange = await _channel.get_exchange(name=_exchange_name, ensure=True) @@ -85,9 +68,10 @@ async def _convert_file_response( 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 - # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute + # 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 + # allow coroutine to execute. calling _future.result() will block the entire thread + # and the coroutine will never execute _channel, _exchange, _routing_key = await await_result( asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop) ) @@ -122,19 +106,22 @@ def is_likely_json(text: str) -> bool: # ============= CleverThisServiceAdapter ============= class CleverThisServiceAdapter: """ - A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the - appropriate CleverThisService API endpoint. - IMPORTANT: This class is not intended to be used directly. It should be subclassed for each CleverThis service - and override the methods to handle the specific API calls. + A base class for CleverThis service adapter that handles the incoming AMQP messages and routes + them to the appropriate CleverThisService API endpoint. + IMPORTANT: This class is not intended to be used directly. It should be subclassed for + each CleverThis service and override the methods to handle the specific API calls. """ def __init__( - self, amq_configuration: AMQConfiguration, session: ClientSession = None + self, + amq_configuration: AMQConfiguration, + routes: List[APIRoute], + 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 - HTTP REST API. + 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 HTTP REST API. """ self.amq_configuration = amq_configuration self.session = session @@ -144,6 +131,79 @@ class CleverThisServiceAdapter: self.require_authenticated_user = ( self.amq_configuration.amq_adapter.require_authenticated_user ) + self.routes = self.group_routes_by_method(api_routes=routes) + + def get_endpoint(self, endpoint: Any) -> Any: + """ + Optionally, Service can override this method to alter the endpoint mapping. + This is useful for example when the CleverThis service has a JSON wrapper + around the endpoint, like CleverSwarm, which might be the preferred endpoint to call. + + :param endpoint: The original endpoint to be called. + :return: The actual endpoint to be called. + """ + return endpoint + + def clone_and_adapt_route(self, original_route: APIRoute) -> APIRoute: + """ + Creates a clone of APIRoute. Checks if there's preferred endpoint with json serialization. + It also adjusts the path regex to ensure it matches the path in different API versions. + """ + + # Need to adapt also the path regex to ensure it matches the path in different API versions + # Ensures the leading '^' is followed by any match and there's no '/' before final '$' + _new_regex_string = "^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$" + _new_route: APIRoute = APIRoute( + path=original_route.path, + endpoint=self.get_endpoint( + original_route.endpoint + ), # Service can alter the endpoint mapping here + response_model=original_route.response_model, + status_code=original_route.status_code, + tags=original_route.tags.copy() if original_route.tags else None, + dependencies=( + original_route.dependencies.copy() if original_route.dependencies else None + ), + summary=original_route.summary, + description=original_route.description, + response_description=original_route.response_description, + responses=(original_route.responses.copy() if original_route.responses else None), + deprecated=original_route.deprecated, + name=original_route.name, + methods=original_route.methods.copy() if original_route.methods else None, + operation_id=original_route.operation_id, + response_model_include=original_route.response_model_include, + response_model_exclude=original_route.response_model_exclude, + response_model_by_alias=original_route.response_model_by_alias, + response_model_exclude_unset=original_route.response_model_exclude_unset, + response_model_exclude_defaults=original_route.response_model_exclude_defaults, + response_model_exclude_none=original_route.response_model_exclude_none, + include_in_schema=original_route.include_in_schema, + response_class=original_route.response_class, + dependency_overrides_provider=original_route.dependency_overrides_provider, + callbacks=(original_route.callbacks.copy() if original_route.callbacks else None), + openapi_extra=( + original_route.openapi_extra.copy() if original_route.openapi_extra else None + ), + generate_unique_id_function=original_route.generate_unique_id_function, + ) + _new_route.path_regex = re.compile(_new_regex_string) + return _new_route + + def group_routes_by_method(self, api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]: + """ + Groups API routes by their HTTP method. This is important to ensure that URLPath is matched + correctly, as different methods may share the same URLPath (e.g. GET, POST PUT, DELETE). + """ + _method_to_routes = defaultdict(list) # Automatically initializes new lists for new keys + for _route in api_routes: + if ( + isinstance(_route, APIRoute) and _route.methods + ): # Ensure methods exist (should always be true for valid routes) + for method in _route.methods: + _method_to_routes[method].append(self.clone_and_adapt_route(_route)) + + return dict(_method_to_routes) def get_content_type(self, message_headers: Dict[str, List[str]]) -> str: """ @@ -158,9 +218,11 @@ class CleverThisServiceAdapter: async def get_current_user(self, token: str) -> object: """ - To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token, - in format appropriate for the service. + To Be Overriden in subclass for given CleverThis Service, to return the current user based + on the token, in format appropriate for the service. + :param token: The token to be used for authentication. + :return: A user object or None if the user is not authenticated. """ return None @@ -174,11 +236,9 @@ class CleverThisServiceAdapter: _auth_user: Any | None = None _amq_response: Any | None = None - logging_debug( - f"on_message: require_authenticated_user={self.require_authenticated_user}" - ) + logging_debug(f"on_message: 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_debug(f"Auth: {_auth}") if isinstance(_auth, List): _auth = _auth[0] @@ -190,12 +250,8 @@ class CleverThisServiceAdapter: 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") - ): - method = message.method.lower() + if _auth_user or not self.require_authenticated_user or message.path().endswith("/login"): + method = message.method().lower() if method == "get": _amq_response = await self.get(message, _auth_user) elif method == "put": @@ -207,117 +263,296 @@ class CleverThisServiceAdapter: elif method == "delete": _amq_response = await self.delete(message, _auth_user) else: - raise ValueError(f"Unexpected HTTP method: {message.method}") + raise ValueError(f"Unexpected HTTP method: {message.method()}") else: _amq_response = DataResponseFactory.of_error_message( - message.id, + message.id(), AMQErrorMessage(401, "Unauthorized", "Unauthorized"), - message.trace_info, + message.trace_info(), ) logging_info( - f"ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}" + f"ALL DONE in on_message id:{message.id()}, resp code:{_amq_response.response_code}" ) return _amq_response - # ================================================================================================== - # ===================== C l e v e r T h i s A P I m e t h o d s ============================ - # ================================================================================================== - - async def get(self, message: DataMessage, auth_user: Any) -> DataResponse: - """ - Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation. - It will attempt to call the REST endpoint directly using HTTP protocol. - :param message: message from the AMQP that contains the GET request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: DataResponse to be sent back to the API Gateway via AMQP - """ - - # 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}" - ) - 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_error_message( - message.id, - AMQErrorMessage(404, "Not Found", "Destination location does not exists"), - message.trace_info, - ) - - async def put(self, message: DataMessage, auth_user: Any) -> DataResponse: - """ - Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation. - It will attempt to call the REST endpoint directly using HTTP protocol. - :param message: message from the AMQP that contains the PUT request - :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._authorize_forward_request( - message, - self.session.put( - f"http://{self.service_host}:{self.service_port}{message.path}", - data=message.body, - headers={**message.headers, **message.trace_info}, - ), - auth_user, - ) - - async def post(self, message: DataMessage, auth_user: Any) -> DataResponse: - """ - Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation. - It will attempt to call the REST endpoint directly using HTTP protocol. - :param message: message from the AMQP that contains the POST request, including body and headers - :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._authorize_forward_request( - message, - self.session.post( - f"http://{self.service_host}:{self.service_port}{message.path}", - data=message.body, - headers={**message.headers, **message.trace_info}, - ), - auth_user, - ) - async def _authorize_forward_request( self, message: DataMessage, request_coroutine, auth_user: Any ) -> DataResponse: """ 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 + Used only for loose / 3rd party mode of coupling with the Service """ # AMQDataParser.parse_request_body(message) + # TODO: implement authorization logic here - ensure the security related headers are set + # This depends on what kind of security is required by the CleverThis service. + # However at this time there is no loosely coupled Service. + return await request_coroutine - async def head(self, message: DataMessage, auth_user: Any) -> DataResponse: + # ============================================================================================== + # ===================== C l e v e r T h i s A P I m e t h o d s ======================== + # ============================================================================================== + + async def get(self, message: AMQMessage, auth_user: Any) -> DataResponse: """ - Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation. - It will attempt to call the REST endpoint directly using HTTP protocol. - :param message: message from the AMQP that contains the POST request, including body and headers + Handles the GET requests for the CleverThis API. In tight coupling (self.session is None), + use path, path params and query string to determine the endpoint method to invoke. + In loose coupling (self.session is not None), forward the request as REST request using + the session, i.e. attempt to call the REST endpoint directly using HTTP protocol. + + :param message: message from the AMQP that contains the GET request :param auth_user: user authenticated by the API Gateway, as UserSchem + :return: DataResponse to be sent back to the API Gateway via AMQP """ + + if self.session is None: + return await self.handle_no_body_payload(message, auth_user, "GET") + + # Forwards as REST - 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}" + ) + _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(), + ) + + async def put(self, message: AMQMessage, auth_user: Any) -> DataResponse: + """ + Handles the PUT requests for the CleverThis API. Uses path, path params, query string and + body payload to determine the endpoint method to invoke and its input arguments + In loose coupling (self.session is not None), attempt to call the REST endpoint directly + using HTTP protocol. + + :param message: message from the AMQP that contains the PUT request + :param auth_user: user authenticated by the API Gateway, as UserSchem + + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + if self.session is None: + return await self.handle_body_payload(message, auth_user, "PUT") + + # Forwards as REST - try to invoke the service directly on this path + return await self._authorize_forward_request( + message, + self.session.put( + f"http://{self.service_host}:{self.service_port}{message.path()}", + data=message.body(), + headers={**message.headers(), **message.trace_info()}, + ), + auth_user, + ) + + async def post(self, message: AMQMessage, auth_user: Any) -> DataResponse: + """ + Handles the POST requests for the CleverThis API. Uses path, path params, query string and + body payload to determine the endpoint method to invoke and its input arguments + In loose coupling (self.session is not None), attempt to call the REST endpoint directly + using HTTP protocol. + + :param message: message from AMQP that contains the POST request, including body and headers + :param auth_user: user authenticated by the API Gateway, as UserSchem + + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + if self.session is None: + return await self.handle_body_payload(message, auth_user, "POST") + + return await self._authorize_forward_request( + message, + self.session.post( + f"http://{self.service_host}:{self.service_port}{message.path()}", + data=message.body(), + headers={**message.headers(), **message.trace_info()}, + ), + auth_user, + ) + + async def head(self, message: AMQMessage, auth_user: Any) -> DataResponse: + """ + Handles the HEAD requests for the CleverThis API. In tight coupling (self.session is None), + use path, path params and query string to determine the endpoint method to invoke. + In loose coupling (self.session is not None), HEAD call is not supported. + + :param message: message from AMQP containing the HEAD request, including body and headers + :param auth_user: user authenticated by the API Gateway, as UserSchema + + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + if self.session is None: + return await self.handle_no_body_payload(message, auth_user, "HEAD") + raise NotImplementedError - async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse: + async def delete(self, message: AMQMessage, auth_user: Any) -> DataResponse: """ - Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation. - It will attempt to call the REST endpoint directly using HTTP protocol. - :param message: message from the AMQP that contains the POST request, including body and headers + Handles the DELETE requests for the CleverThis API. + + :param message: message from AMQP that contains the DELETE request, including body & headers + :param auth_user: user authenticated by the API Gateway, as UserSchema + + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + if self.session is None: + return await self.handle_no_body_payload(message, auth_user, "DELETE") + + return await self._authorize_forward_request( + message, + self.session.post( + f"http://{self.service_host}:{self.service_port}{message.path()}", + data=message.body(), + headers={**message.headers(), **message.trace_info()}, + ), + auth_user, + ) + + async def options(self, message: AMQMessage, auth_user: Any) -> DataResponse: + """ + Handles the OPTIONS requests. In tight coupling (self.session is None), + use path, path params and query string to determine the endpoint method to invoke. + In loose coupling (self.session is not None), forward the request as REST request using + the session, i.e. attempt to call the REST endpoint directly using HTTP protocol. + + :param message: message from AMQP that contains the OPTIONS request, including headers + :param auth_user: user authenticated by the API Gateway, as UserSchema + + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + if self.session is None: + return await self.handle_no_body_payload(message, auth_user, "OPTIONS") + + raise NotImplementedError + + async def patch(self, message: AMQMessage, auth_user: Any) -> DataResponse: + """ + Handles the PATCH requests for the CleverBRAG API. + + :param message: message from AMQP that contains the PATCH request, including body & headers + :param auth_user: user authenticated by the API Gateway, as UserSchema + + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + if self.session is None: + return await self.handle_no_body_payload(message, auth_user, "PATCH") + + return await self._authorize_forward_request( + message, + self.session.post( + f"http://{self.service_host}:{self.service_port}{message.path()}", + data=message.body(), + headers={**message.headers(), **message.trace_info()}, + ), + auth_user, + ) + + # ============================================================================================== + # ================= C l e v e r S w a r m p a y l o a d h a n d l e r s ================== + # ============================================================================================== + + async def handle_no_body_payload( + self, message: AMQMessage, auth_user: Any, method: str + ) -> DataResponse: + """ + Handles the requests that do not have a body payload. + :param message: message from the AMQP that contains the request :param auth_user: user authenticated by the API Gateway, as UserSchem :return: DataResponse to be sent back to the API Gateway via AMQP """ - raise NotImplementedError + _routes = self.routes.get(method, []) + _path, _args = AMQDataParser.parse_get_url(message.path()) + for _route in _routes: + _match = _route.path_regex.match(_path) + if _match: + _args.update(_match.groupdict()) + _args["authenticated_user"] = auth_user + # If the path matches, call the corresponding function + return await CleverThisServiceAdapter.call_cleverthis_api( + message, + _args, + api_method=_route.endpoint, + success_code=_route.status_code, + loop=self.loop, + ) + + return DataResponseFactory.of( + message.id(), + 404, + "text/plain", + str(f"Destination location [{message.path()}] does not exists").encode("utf-8"), + message.trace_info(), + ) + + async def handle_body_payload( + self, message: AMQMessage, auth_user: Any, method: str + ) -> DataResponse: + """ + Handles the requests that have a body payload, including Multipart data. + :param message: message from the AMQP that contains the request + :param auth_user: user authenticated by the API Gateway, as UserSchem + :param method: HTTP method (POST, PUT, etc.) + :return: DataResponse to be sent back to the API Gateway via AMQP + """ + _form_data = AMQDataParser.parse_request_body(message) + _routes = self.routes.get(method, []) + _path, _query_args = AMQDataParser.parse_get_url(message.path()) + for _route in _routes: + _match = _route.path_regex.match(_path) + if _match: + # If the path matches, call the corresponding function + _form_data.update(_match.groupdict()) + _form_data.update(_query_args) + for _body_param in _route.dependant.body_params: + if _body_param.name in _form_data: + _file_data = _form_data[_body_param.name] + actual_filepath = message.extra_data[_body_param.name] + _form_data[_body_param.name] = ( + _body_param.type_.__name__ == "UploadFile" + and UploadFile( + file=pathlib.Path(actual_filepath).open("rb"), + size=_file_data["size"], + filename=_file_data["filename"], + headers=Headers({"Content-Type": _file_data["mediaType"]}), + ) + or _form_data[_body_param.name] + ) + + _verified_args = {} + _required_args = getattr(_route.endpoint, "__annotations__", {}) + for arg in _required_args: + if arg in _form_data: + _verified_args[arg] = _form_data[arg] + else: + # needed for /login input + if arg == "authenticated_user": + _verified_args["authenticated_user"] = auth_user + if _required_args[arg].__name__ == "OAuth2PasswordRequestForm": + try: + _verified_args[arg] = OAuth2PasswordRequestForm(**_form_data) + except Exception as e: + print(f"Error: {e}") + + return await CleverThisServiceAdapter.call_cleverthis_api( + message, + _verified_args, + api_method=_route.endpoint, + success_code=_route.status_code if _route.status_code else 200, + loop=self.loop, + ) + + return DataResponseFactory.of( + message.id(), + 404, + "text/plain", + str(f"Destination location [{message.path()}] does not exists").encode("utf-8"), + message.trace_info(), + ) @staticmethod async def call_cleverthis_api( @@ -328,13 +563,15 @@ class CleverThisServiceAdapter: loop: AbstractEventLoop | None = None, ) -> DataResponse: """ - Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary. - Invokes the provided API method with the given arguments and authenticated user, - and returns the response as DataResponse. + Handles POST and PUT requests, which typically contain possibly complex payload data, + passed as a dictionary. Invokes the provided API method with the given arguments and + authenticated user, and returns the response as DataResponse. + :param message: DataMessage containing the details of the call and the payload :param args: map of the arguments to pass to the API method :param api_method: Callable - the method to invoke. Last parameter is the authenticated user :param success_code: HTTP code to be returned when operation suceeds + :return: DataResponse class """ try: @@ -350,23 +587,19 @@ class CleverThisServiceAdapter: elif arg.annotation == float: _verified_args[arg.name] = float(args[arg.name]) elif arg.annotation == bool: - _verified_args[arg.name] = serializer.str_to_bool( - args[arg.name] - ) + _verified_args[arg.name] = serializer.str_to_bool(args[arg.name]) # TODO: more? else: _verified_args[arg.name] = args[arg.name] 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())}" + "INVOKE ENDPOINT: {%s}, ARGS: {%s}", + api_method.__name__, + ", ".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 = ( @@ -391,60 +624,56 @@ class CleverThisServiceAdapter: else ( _json_result.body if hasattr(_json_result, "body") - else ( - _json_result.read() - if hasattr(_json_result, "read") - else _json_result - ) + else (_json_result.read() if hasattr(_json_result, "read") else _json_result) ) ) logging_info( - f"ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}" + f"ALL DONE in call_cleverthis_api id:{message.id()}, resp code:{success_code}" ) return DataResponseFactory.of( - message.id, + message.id(), success_code, _content_type, _binary_result, - message.trace_info, + message.trace_info(), ) except HTTPException as http_error: logging_error(f"Service REST endpoint invocation failed: {http_error}") _content = str(http_error.detail) if not is_likely_json(_content): return DataResponseFactory.of_error_message( - message.id, + message.id(), AMQErrorMessage( http_error.status_code, "Service Invocation Failed", _content, ), - message.trace_info, + message.trace_info(), ) return DataResponseFactory.of( - message.id, + message.id(), http_error.status_code, "application/json", _content.encode("utf-8"), - message.trace_info, + message.trace_info(), ) except Exception as error: logging_error(f"Service endpoint invocation failed: {error}") _content = str(error) if not is_likely_json(_content): return DataResponseFactory.of_error_message( - message.id, + message.id(), AMQErrorMessage( 500, "Service Invocation Failed", _content, ), - message.trace_info, + message.trace_info(), ) return DataResponseFactory.of( - message.id, + message.id(), 500, "application/json", _content.encode("utf-8"), - message.trace_info, + message.trace_info(), ) diff --git a/amqp/adapter/data_message_factory.py b/amqp/adapter/data_message_factory.py index 2d10e88..b42f011 100644 --- a/amqp/adapter/data_message_factory.py +++ b/amqp/adapter/data_message_factory.py @@ -5,21 +5,21 @@ methods to create, serialize and deserialize DataMessage objects. import base64 import json import re -from io import BytesIO from datetime import datetime, timezone +from io import BytesIO from typing import Dict, List import amqp.adapter.file_handler -from amqp.model.model import DataMessage -from amqp.model.snowflake_id import SnowflakeId +from amqp.adapter.file_handler import FileHandler, StreamingFileHandler from amqp.adapter.serializer import ( - map_with_list_as_string, map_as_string, + map_with_list_as_string, parse_map_list_string, parse_map_string, ) +from amqp.model.model import DataMessage, DataMessageMagicByte +from amqp.model.snowflake_id import SnowflakeId from amqp.router.utils import sanitize_as_rabbitmq_name -from amqp.adapter.file_handler import StreamingFileHandler, FileHandler class DataMessageFactory: @@ -30,7 +30,6 @@ class DataMessageFactory: MACHINE_ID_MASK = (1 << MACHINE_ID_BITS) - 1 SEQUENCE_MASK = (1 << SEQUENCE_BITS) - 1 - MAGIC_BYTE_HTTP_REQUEST = "A" _instance = None snowflake_epoch = datetime(2024, 10, 1, 0, 0, tzinfo=timezone.utc).timestamp() @@ -100,12 +99,10 @@ 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, + DataMessageMagicByte.HTTP_REQUEST.value, self.generate_snowflake_id(), method, domain, @@ -122,7 +119,7 @@ class DataMessageFactory: :param message: DataMessage input :return: routing key (as string) compliant to RabbitMQ allowed character set. """ - return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}") + return sanitize_as_rabbitmq_name(f"{message.domain()}.{message.path()}") @staticmethod def get_timestamp_from_id(_id: SnowflakeId) -> int: @@ -132,9 +129,7 @@ class DataMessageFactory: :return: the message timestamp """ bits = _id.get_bits() - low = bits[0] >> ( - 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 ) @@ -149,13 +144,13 @@ class DataMessageFactory: """ return ( f"DataMessage{{" - f"id={message.id}, " - f"method='{message.method}', " - f"domain='{message.domain}', " - f"path='{message.path}', " - f"headers={{{map_with_list_as_string(message.headers)}}}, " - f"traceInfo={{{map_as_string(message.trace_info)}}}, " - f"base64body={message.base64_body.decode()}" + f"id={message.id()}, " + f"method='{message.method()}', " + f"domain='{message.domain()}', " + f"path='{message.path()}', " + f"headers={{{map_with_list_as_string(message.headers())}}}, " + f"traceInfo={{{map_as_string(message.trace_info())}}}, " + f"base64body={message.base64body().decode()}" f"}}" ) @@ -166,19 +161,19 @@ class DataMessageFactory: :param message: message to serialize :return: messages as bytes """ - id_bytes = str(message.id).encode() + 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=" + 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") header_length = 1 + len(id_bytes) + len(prolog) prolog_len = header_length.to_bytes(2, byteorder="big") - msg_length = 2 + header_length + len(message.base64_body) + msg_length = 2 + header_length + len(message.base64body()) with BytesIO(bytearray(msg_length)) as baos: baos.write(prolog_len) - baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8")) + baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8")) baos.write(id_bytes) baos.write(prolog) - baos.write(message.base64_body) + baos.write(message.base64body()) return baos.getvalue() @staticmethod @@ -192,7 +187,7 @@ class DataMessageFactory: metadata = input_bytes[2 : prolog_len + 2].decode() pattern = re.compile( - f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}" + f"{DataMessageMagicByte.HTTP_REQUEST.value}" r"([0-9A-Fa-f]+)\|" r"m=([^|]*)\|" r"d=([^|]*)\|" @@ -220,7 +215,7 @@ class DataMessageFactory: trace_info = parse_map_string(trace_info_str) return DataMessage( - DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST, + DataMessageMagicByte.HTTP_REQUEST.value, SnowflakeId.from_hex(id_str), method, domain, diff --git a/amqp/adapter/data_parser.py b/amqp/adapter/data_parser.py index 7ba9425..47f23fa 100644 --- a/amqp/adapter/data_parser.py +++ b/amqp/adapter/data_parser.py @@ -1,16 +1,15 @@ import io import json -import python_multipart -from tempfile import SpooledTemporaryFile - -from fastapi import UploadFile -from python_multipart.multipart import Field, File -from typing import Tuple, Dict +from typing import Dict, Tuple from urllib.parse import parse_qs, urlparse from xml.etree import ElementTree -from amqp.adapter.logging_utils import logging_error, logging_debug -from amqp.model.model import DataMessage +import python_multipart +from fastapi import UploadFile +from python_multipart.multipart import Field, File + +from amqp.adapter.logging_utils import logging_debug, logging_error +from amqp.model.model import AMQMessage class MultipartFormDataParser: @@ -23,15 +22,14 @@ class MultipartFormDataParser: included, the message is transmitted in the unchanged multipart/form-data format. """ - def __init__(self, message: DataMessage): + def __init__(self, message: AMQMessage): self.form_data: dict = {} self.is_multipart = False 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() + key: "\n".join(value).encode("utf-8") for key, value in message.headers().items() } try: if ( @@ -53,9 +51,7 @@ class MultipartFormDataParser: ) 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 @@ -72,11 +68,8 @@ class MultipartFormDataParser: def on_file(self, file: File): logging_debug(str(file)) file.file_object.seek(0) - _spooled_file: SpooledTemporaryFile = SpooledTemporaryFile() - _spooled_file.write(file.file_object.read()) - _spooled_file.seek(0) self.form_data[file.field_name.decode("utf-8")] = UploadFile( - _spooled_file, size=file.size, filename=file.file_name.decode("utf-8") + file.file_object, size=file.size, filename=file.file_name.decode("utf-8") ) @@ -135,13 +128,11 @@ class AMQDataParser: _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 @staticmethod - def parse_request_body(message: DataMessage): + def parse_request_body(message: AMQMessage): """ Parses the HTTP POST request body based on the MIME type. This is typically called fom the concrete CleverThis Service to extract parameters from the message body. @@ -151,7 +142,7 @@ class AMQDataParser: 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] @@ -167,15 +158,12 @@ class AMQDataParser: 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() + 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"]: - _form_data = MultipartFormDataParser.process_form_data( - _form_data - ) + _form_data = AMQDataParser.process_form_data(_form_data) return _form_data except Exception as e: logging_error(f"Invalid URL-encoded form data: {e}") diff --git a/amqp/adapter/data_response_factory.py b/amqp/adapter/data_response_factory.py index 5ee7908..96e0c39 100644 --- a/amqp/adapter/data_response_factory.py +++ b/amqp/adapter/data_response_factory.py @@ -5,13 +5,13 @@ method to create, serialize and deserialize DataResponse message import base64 import re from io import BytesIO -from amqp.model.model import DataResponse, AMQErrorMessage + +from amqp.adapter.serializer import map_as_string, parse_map_string +from amqp.model.model import AMQErrorMessage, DataMessageMagicByte, DataResponse from amqp.model.snowflake_id import SnowflakeId -from amqp.adapter.serializer import parse_map_string, map_as_string class DataResponseFactory: - MAGIC_BYTE_HTTP_REQUEST = "A" @staticmethod def create_async_response_message(_id: SnowflakeId) -> DataResponse: @@ -20,9 +20,7 @@ class DataResponseFactory: :param _id: ID :return: Response message """ - return DataResponse( - str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {} - ) + return DataResponse(str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {}) @staticmethod def of( @@ -76,19 +74,19 @@ 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=" + 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") header_length = 1 + len(id_bytes) + len(prolog) prolog_len = header_length.to_bytes(2, byteorder="big") - msg_length = 2 + header_length + len(response.response) + msg_length = 2 + header_length + len(response.base64body()) with BytesIO(bytearray(msg_length)) as baos: baos.write(prolog_len) - baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8")) + baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8")) baos.write(id_bytes) baos.write(prolog) - baos.write(base64.b64encode(response.response)) + baos.write(base64.b64encode(response.base64body())) return baos.getvalue() @staticmethod @@ -102,7 +100,7 @@ class DataResponseFactory: metadata = input_bytes[2 : prolog_len + 2].decode() pattern = re.compile( - f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}" + f"{DataMessageMagicByte.HTTP_REQUEST.value}" r"([0-9A-Fa-f]+)\|" r"r=([^|]*)\|" r"c=([^|]*)\|" diff --git a/amqp/adapter/file_handler.py b/amqp/adapter/file_handler.py index a2eacf9..7c3e0bf 100644 --- a/amqp/adapter/file_handler.py +++ b/amqp/adapter/file_handler.py @@ -191,9 +191,7 @@ class StreamingFileHandler(FileHandler): _eof = IN_PROGRESS channel: AbstractRobustChannel = await connection.channel() queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False) - logging_debug( - f"Awaiting all file chunks being downloaded for {filename}" - ) + logging_debug(f"Awaiting all file chunks being downloaded for {filename}") async with queue.iterator() as queue_iter: async for message in queue_iter: async with message.process(): @@ -236,12 +234,8 @@ class StreamingFileHandler(FileHandler): _res = bytearray() try: channel: AbstractRobustChannel = await _connection.channel() - queue: AbstractQueue = await channel.get_queue( - queue_name, ensure=False - ) - logging_debug( - f"Awaiting all buffer chunks being downloaded for {queue_name}" - ) + queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False) + logging_debug(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(): @@ -296,22 +290,16 @@ class StreamingFileHandler(FileHandler): 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 defaultdict() # 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}" - ) + logging_info(f" [{message.delivery_tag}] about to create task for {file_info}") _task = asyncio.create_task( - self.download_file( - loop, rabbitmq_url, file_info, _message_data, output_dir - ) + self.download_file(loop, rabbitmq_url, file_info, _message_data, output_dir) ) _tasks.append(_task) logging_info( @@ -377,9 +365,7 @@ class StreamingFileHandler(FileHandler): "total_size": _file_size, "chunk_size": len(_chunk), "eof": ( - END_OF_FILE - if max_chunk_size + _offset >= _file_size - else IN_PROGRESS + END_OF_FILE if max_chunk_size + _offset >= _file_size else IN_PROGRESS ), }, ) @@ -392,9 +378,7 @@ class StreamingFileHandler(FileHandler): 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( diff --git a/amqp/adapter/pydantic_serializer.py b/amqp/adapter/pydantic_serializer.py index 6bee309..1fdbf72 100644 --- a/amqp/adapter/pydantic_serializer.py +++ b/amqp/adapter/pydantic_serializer.py @@ -2,12 +2,13 @@ This module provides serialization and deserialization functions for Pydantic models that extend BaseModel """ -import uuid import json +import uuid from datetime import datetime -from typing import Dict, Any +from typing import Any, Dict +from urllib.parse import parse_qs, urlparse + from pydantic import BaseModel -from urllib.parse import urlparse, parse_qs from amqp.adapter.logging_utils import logging_debug diff --git a/amqp/adapter/serializer.py b/amqp/adapter/serializer.py index 86965f3..a112507 100644 --- a/amqp/adapter/serializer.py +++ b/amqp/adapter/serializer.py @@ -4,7 +4,7 @@ Helper functions used to aid with (de)serialization of messages sent over AMQP. import struct from io import BytesIO -from typing import List, Union, Dict +from typing import Dict, List, Union def long_to_bytes(x: int) -> bytes: diff --git a/amqp/adapter/service_message_factory.py b/amqp/adapter/service_message_factory.py index 4d0c797..2e03d74 100644 --- a/amqp/adapter/service_message_factory.py +++ b/amqp/adapter/service_message_factory.py @@ -98,9 +98,7 @@ 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( @@ -110,11 +108,7 @@ 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( @@ -147,9 +141,7 @@ class ServiceMessageFactory: :return: formatted message type """ num_type = ( - message.message_type.value - if message.message_type - else ServiceMessageType.UNKNOWN.value + 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/config/amq_configuration.py b/amqp/config/amq_configuration.py index 905e356..383c7ef 100644 --- a/amqp/config/amq_configuration.py +++ b/amqp/config/amq_configuration.py @@ -41,9 +41,7 @@ class AMQAdapter: fallback=False, ) # these 2 shall be provided by Docker swarm, format is a random alphanumeric string - self.swarm_service_id = os.getenv( - "SWARM_SERVICE_ID", default="clever-amqp-service" - ) + self.swarm_service_id = os.getenv("SWARM_SERVICE_ID", default="clever-amqp-service") self.swarm_task_id = os.getenv("SWARM_TASK_ID", default="python-adapter") def adapter_prefix(self) -> str: @@ -68,15 +66,9 @@ class Dispatch: :param config: config parser to read configuration values """ - self.use_dlq = config.getboolean( - "CleverMicro-AMQ", self.PREFIX + "use-dlq", 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.use_dlq = config.getboolean("CleverMicro-AMQ", self.PREFIX + "use-dlq", 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 ) diff --git a/amqp/healthcheck.py b/amqp/healthcheck.py index fb86847..b1038a0 100644 --- a/amqp/healthcheck.py +++ b/amqp/healthcheck.py @@ -1,4 +1,5 @@ import sys + import requests url = "http://localhost:8080/amq-adapter-healthcheck" diff --git a/amqp/model/model.py b/amqp/model/model.py index 9ee4bee..8037576 100644 --- a/amqp/model/model.py +++ b/amqp/model/model.py @@ -4,6 +4,8 @@ from dataclasses import dataclass, field from enum import Enum from typing import Dict, List +from aio_pika.abc import AbstractRobustConnection + from amqp.model.service_message_type import ServiceMessageType from amqp.model.snowflake_id import SnowflakeId from amqp.router.utils import sanitize_as_rabbitmq_name @@ -16,6 +18,11 @@ Define the structure of messages used in Clever Micro """ +class DataMessageMagicByte(Enum): + HTTP_REQUEST = "A" + RPC_REQUEST = "R" + + @dataclass class CleverMicroMessage: DEFAULT_CONTENT_TYPE = ["application/octet-stream"] @@ -105,9 +112,7 @@ class DataMessage(CleverMicroMessage): trace_info: Dict[str, str], base64body: bytes, ): - super().__init__( - magic, id, method, domain, path, headers, trace_info, base64body - ) + super().__init__(magic, id, method, domain, path, headers, trace_info, base64body) def method(self) -> str: return self.m_field() @@ -122,6 +127,51 @@ class DataMessage(CleverMicroMessage): return sanitize_as_rabbitmq_name(f"{self.d_field()}.{self.p_field()}") +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.base64body(), + ) + self.extra_data = extra_data + + +@dataclass +class AMQMessage(DataMessage): + """ + Expand the core DataMessage type to include the channel, which is required to create a dedicated queue + in case the response should initiate file download. + """ + + _connection: AbstractRobustConnection = field(repr=False) # Don't show in repr + + def __init__(self, data_message: DataMessage, connection: AbstractRobustConnection): + # Initialize the parent DataMessage with all properties from the input data_message + super().__init__( + magic=data_message.magic(), + id=data_message.id(), + method=data_message.method(), + domain=data_message.domain(), + path=data_message.path(), + headers=data_message.headers(), + trace_info=data_message.trace_info(), + base64body=data_message.base64body(), + ) + self._connection = connection + self.extra_data = data_message.extra_data if hasattr(data_message, "extra_data") else {} + + @property + def connection(self) -> AbstractRobustConnection: + """Get the AMQP connection associated with this message""" + return self._connection + + class DataResponse(CleverMicroMessage): """ The response message to REST & RPC style calls (note: the request is made with DataMessage) @@ -211,21 +261,6 @@ class AMQRoute: return self.__str__() -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, - ) - self.extra_data = extra_data - - # Ensure backward compatibility AMQResponse = DataResponse @@ -254,7 +289,7 @@ class ScalingRequestAlert(Enum): IDLE = "IDLE" UPDATE = "UPDATE" SCALE_UP = "SCALE_UP" - SCSLE_DOWN = "SCALE_DOWN" + SCALE_DOWN = "SCALE_DOWN" @dataclass(frozen=True) @@ -289,9 +324,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/rabbitmq/rabbit_mq_client.py b/amqp/rabbitmq/rabbit_mq_client.py index 0c3ca58..3804a84 100644 --- a/amqp/rabbitmq/rabbit_mq_client.py +++ b/amqp/rabbitmq/rabbit_mq_client.py @@ -1,21 +1,21 @@ import logging import aio_pika -from aio_pika.abc import AbstractRobustQueue, AbstractRobustExchange +from aio_pika.abc import AbstractRobustExchange, AbstractRobustQueue from pika.exchange_type import ExchangeType from amqp.adapter.data_message_factory import DataMessageFactory from amqp.adapter.logging_utils import ( - logging_error, logging_debug, + logging_error, logging_info, logging_warning, ) -from amqp.model.model import DataMessage, AMQRoute +from amqp.model.model import AMQRoute, DataMessage from amqp.router.router_base import ( AMQ_REPLY_TO_EXCHANGE, - DLQ_EXCHANGE, AMQ_RPC_EXCHANGE, + DLQ_EXCHANGE, ) from amqp.router.router_consumer import RouterConsumer from amqp.router.utils import sanitize_as_rabbitmq_name @@ -85,21 +85,15 @@ 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() + 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( @@ -109,13 +103,9 @@ class RabbitMQClient: 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 - ) + _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) @@ -200,16 +190,14 @@ class RabbitMQClient: message ), # context path is a routing key ) - logging_info( - f"Msg Publish (no-reply), messageId: {message.id}, to: {route}" - ) + logging_info(f"Msg Publish (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", delivery_mode=2, - correlation_id=str(message.id), + correlation_id=str(message.id()), reply_to=self.router.get_reply_to_queue_name(), ) await self.rpc_exchange.publish( @@ -220,12 +208,14 @@ class RabbitMQClient: ) logging_info( "Msg Publish (RPC call), messageId: %s, to: %s", - message.id.as_string(), + message.id().as_string(), route.as_string(), ) else: logging_warning( - "Message not sent, no route for %s/%s", message.domain, message.path + "Message not sent, no route for %s/%s", + message.domain(), + message.path(), ) except Exception as err: diff --git a/amqp/router/route_database.py b/amqp/router/route_database.py index 637b793..4f36d68 100644 --- a/amqp/router/route_database.py +++ b/amqp/router/route_database.py @@ -40,13 +40,9 @@ 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 a683db1..dd760f1 100644 --- a/amqp/router/router_base.py +++ b/amqp/router/router_base.py @@ -27,7 +27,7 @@ class RouterBase: self.route_removed_notifier: Callable[[AMQRoute], None] | None = None def get_routing_key(self, message: DataMessage) -> str: - return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}") + return sanitize_as_rabbitmq_name(f"{message.domain()}.{message.path()}") def wrap_routes(self) -> str: return self.route_database.wrap_routes() @@ -37,10 +37,7 @@ class RouterBase: def unwrap_route_list(self, route_string: str) -> List[AMQRoute]: logging_debug("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]: @@ -50,14 +47,12 @@ class RouterBase: return self.route_database.find_route(domain, path) def find_route_by_message(self, message: DataMessage) -> AMQRoute | None: - return self.route_database.find_route(message.domain, message.path) + return self.route_database.find_route(message.domain(), message.path()) 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: @@ -81,14 +76,9 @@ 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_error(f"Can't remove route(s): {err}") diff --git a/amqp/router/router_consumer.py b/amqp/router/router_consumer.py index 23384b8..10f0a23 100644 --- a/amqp/router/router_consumer.py +++ b/amqp/router/router_consumer.py @@ -77,17 +77,13 @@ class RouterConsumer(RouterProducer): 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__() @@ -101,9 +97,7 @@ class RouterConsumer(RouterProducer): 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_debug( "******[%s] (ROUTING_DATA_REQ) msg=%s, routeMap=[%s]", @@ -124,9 +118,7 @@ class RouterConsumer(RouterProducer): 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", @@ -158,9 +150,7 @@ class RouterConsumer(RouterProducer): 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("RouterConsumer registered Route: %s", route) else: @@ -182,9 +172,7 @@ class RouterConsumer(RouterProducer): service_message: ServiceMessage = self.service_message_factory.of( ServiceMessageType.ROUTING_DATA_REMOVE, route.as_string(), ANY_RECIPIENT ) - if not await self.publish_service_message( - service_message, self.cm_response_exchange - ): + if not await self.publish_service_message(service_message, self.cm_response_exchange): logging_warning( "RouterConsumer couldn't remove current route: %s, channel not open.", route, @@ -201,13 +189,9 @@ class RouterConsumer(RouterProducer): * Cleanup - de-register routes with master. """ routes: Set[AMQRoute] = self.get_routes() - logging_info( - "RouterConsumer.cleanup() [%d]: %s", len(routes), self.wrap_routes() - ) + logging_info("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 + 1 for r in (self.remove_route_on_cleanup(route) for route in routes) if not r ) if not_removed > 0: logging_warning( diff --git a/amqp/router/router_producer.py b/amqp/router/router_producer.py index 054a541..6224dc1 100644 --- a/amqp/router/router_producer.py +++ b/amqp/router/router_producer.py @@ -47,9 +47,7 @@ class RouterProducer(RouterBase): """ 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 @@ -98,9 +96,7 @@ class RouterProducer(RouterBase): 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 @@ -114,9 +110,7 @@ class RouterProducer(RouterBase): 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): """ @@ -130,9 +124,7 @@ class RouterProducer(RouterBase): message.message_id, ) 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_debug( "******[%s] ROUTING_DATA, msg=%s, routeMap=[%s] sn=%s, replyTo=%s", @@ -144,8 +136,7 @@ class RouterProducer(RouterBase): ) if ( cm_message.is_valid() - and not self.amq_configuration.amq_adapter.service_name - == cm_message.reply_to + 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) @@ -164,9 +155,7 @@ class RouterProducer(RouterBase): 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 - ): + if not await self.publish_service_message(serviceMessage, self.cm_request_exchange): logging_warning( "RouterProducer could not request route data: %s channel is not open.", CM_REQUEST_EXCHANGE, diff --git a/amqp/service/amq_message_handler.py b/amqp/service/amq_message_handler.py index c099e96..af67d0d 100644 --- a/amqp/service/amq_message_handler.py +++ b/amqp/service/amq_message_handler.py @@ -22,6 +22,7 @@ from amqp.adapter.service_message_factory import ServiceMessageFactory from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import ( DataMessage, + DataMessageMagicByte, DataResponse, MultipartDataMessage, ) @@ -74,21 +75,15 @@ class DataMessageHandler: self.file_handler: FileHandler = file_handler self.backpressure_handler: BackpressureHandler = backpressure_handler - async def inbound_data_message_callback_as_thread( - self, message: AbstractIncomingMessage - ): + async def inbound_data_message_callback_as_thread(self, message: AbstractIncomingMessage): """ The callback method that is called when a DataMessage is received from RabbitMQ, starts new Thread. """ threading.Thread( - target=lambda: asyncio.run( - self.inbound_data_message_callback_no_thread(message) - ) + target=lambda: asyncio.run(self.inbound_data_message_callback_no_thread(message)) ).start() - async def inbound_data_message_callback_no_thread( - self, message: AbstractIncomingMessage - ): + async def inbound_data_message_callback_no_thread(self, message: AbstractIncomingMessage): """ Receives the DataMessage as bytes from RabbitMQ queue and deserializes it to the DataMessage object Ensures Jaeger trace-id propagation. @@ -100,19 +95,17 @@ class DataMessageHandler: amq_message = await self.reconstitute_data_message(message) if amq_message is not None: self.ensure_trace_info(amq_message) - if amq_message.magic == DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST: + if amq_message.magic() == DataMessageMagicByte.HTTP_REQUEST.value: 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, + 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) @@ -176,9 +169,7 @@ class DataMessageHandler: amq_message = await DataMessageFactory.from_stream( message.body, rabbitmq_url=self.rabbit_mq_url, loop=self.loop ) - logging_info( - f"######[{delivery_tag}] AMQ Message from stream: {amq_message}" - ) + logging_info(f"######[{delivery_tag}] AMQ Message from stream: {amq_message}") if amq_message is not None: extra_data = await self.file_handler.on_message( message=message, @@ -187,26 +178,24 @@ class DataMessageHandler: loop=self.loop, output_dir=self.output_dir, ) - logging_info( - f"######[{delivery_tag}] Multipart download result: {extra_data}" - ) + logging_info(f"######[{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), + 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() + 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) @@ -215,7 +204,7 @@ class DataMessageHandler: # logging.info(" [*] CTX2=%s", context_with_span) # propagator.inject(context_with_span, amq_message.trace_info, self.set_text) # logging.info(" [***] NEW TI=%s CTX=%s SCOPE=%s", - # map_as_string(amq_message.trace_info), context_with_span, context) + # map_as_string(amq_message.trace_info()), context_with_span, context) def record_duration(self, start, delivery_tag): global last_data_message_time @@ -237,17 +226,15 @@ class DataMessageHandler: if reply_to: pika_message: Message = Message( body=DataResponseFactory.serialize(response), - correlation_id=response.id, + correlation_id=response.id(), content_type=( - response.content_type[0] - if isinstance(response.content_type, list) - else response.content_type + response.content_type()[0] + if isinstance(response.content_type(), list) + else response.content_type() ), ) _res = asyncio.run_coroutine_threadsafe( - self.reply_to_exchange.publish( - message=pika_message, routing_key=reply_to - ), + self.reply_to_exchange.publish(message=pika_message, routing_key=reply_to), self.loop, ) logging_debug( @@ -270,9 +257,7 @@ class DataMessageHandler: logging_debug(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 d7e583d..3494880 100644 --- a/amqp/service/amq_service.py +++ b/amqp/service/amq_service.py @@ -1,26 +1,24 @@ +import asyncio import logging import logging.config import os -import asyncio from asyncio import Future from threading import Thread from typing import Optional from aio_pika.abc import AbstractRobustExchange +from amqp.adapter.amq_route_factory import AMQRouteFactory from amqp.adapter.backpressure_handler import BackpressureHandler from amqp.adapter.data_message_factory import DataMessageFactory -from amqp.adapter.amq_route_factory import AMQRouteFactory from amqp.adapter.logging_utils import logging_info, logging_warning from amqp.adapter.service_message_factory import ServiceMessageFactory - from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import DataMessage from amqp.rabbitmq.rabbit_mq_client import RabbitMQClient from amqp.service.amq_message_handler import DataMessageHandler 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") @@ -66,9 +64,7 @@ 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) backpressure_handler = BackpressureHandler( channel=self.rabbit_mq_client.channel, loop=loop, @@ -122,7 +118,7 @@ class AMQService: return None def send_message(self, message: DataMessage, future: Future): - self.amq_data_message_handler.outstanding[str(message.id)] = future + self.amq_data_message_handler.outstanding[str(message.id())] = future def remove_outstanding_future(self, message_id: str): self.amq_data_message_handler.outstanding.pop(message_id, None) diff --git a/amqp/service/tracing.py b/amqp/service/tracing.py index 4d4210f..bb0fac6 100644 --- a/amqp/service/tracing.py +++ b/amqp/service/tracing.py @@ -1,15 +1,14 @@ -from opentelemetry import metrics, trace -from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.trace import TracerProvider, Tracer -from opentelemetry.sdk.trace.export import BatchSpanProcessor -from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter - -from prometheus_client import start_http_server -from opentelemetry.exporter.prometheus import PrometheusMetricReader -from opentelemetry.sdk.resources import SERVICE_NAME, Resource - import os +from opentelemetry import metrics, trace +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.exporter.prometheus import PrometheusMetricReader +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from opentelemetry.sdk.trace import Tracer, TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from prometheus_client import start_http_server + def initialize_trace(app=None, name=None) -> Tracer: # Resource can be required for some backends, e.g. Jaeger or Prometheus @@ -17,9 +16,7 @@ def initialize_trace(app=None, name=None) -> Tracer: # 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") - ) + SERVICE_NAME: (name if name else os.environ.get("APPLICATION_NAME", "Python App")) } ) # diff --git a/cleverbrag/clever_brag_adapter.py b/cleverbrag/clever_brag_adapter.py index 65d847c..47112b7 100644 --- a/cleverbrag/clever_brag_adapter.py +++ b/cleverbrag/clever_brag_adapter.py @@ -4,101 +4,23 @@ The code IS MEANT to be part of CleverBRAG codebase, and has direct dependencies It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the CleverBRAG context. """ -import asyncio -import re -from collections import defaultdict from threading import Thread -from typing import List, Dict, Any -from fastapi.routing import APIRoute -from fastapi.security import OAuth2PasswordRequestForm -from fastapi.security.http import HTTPAuthorizationCredentials -from pydantic import BaseModel +from typing import Any, List from uuid import UUID -from starlette.datastructures import UploadFile, Headers -import pathlib # CleverBrag specific imports import core -from core import R2RConfig, R2RBuilder, R2RProviderFactory -from sdk.async_client import create_r2r_app +from fastapi.routing import APIRoute +from fastapi.security.http import HTTPAuthorizationCredentials +from pydantic import BaseModel from shared.abstractions import R2RSerializable, User # AMQP specific imports -from amqp.adapter.data_response_factory import AMQResponseFactory -from amqp.adapter.data_parser import AMQDataParser -from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage +from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import AMQResponse from amqp.service.amq_service import AMQService -def clone_and_adapt_route(original_route: APIRoute) -> APIRoute: - """ - Creates a clone of an APIRoute. It checks if there's preferred endpoint with json serialization. - It also adjusts the path regex to ensure it matches the path in different API versions. - """ - - # Need to adapt also the path regex to ensure it matches the path in different API versions - # Ensures the leading '^' is followed by any match and there's no '/' before final '$' - _new_regex_string = ( - "^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$" - ) - _new_route: APIRoute = APIRoute( - path=original_route.path, - endpoint=original_route.endpoint, - response_model=original_route.response_model, - status_code=original_route.status_code, - tags=original_route.tags.copy() if original_route.tags else None, - dependencies=( - original_route.dependencies.copy() if original_route.dependencies else None - ), - summary=original_route.summary, - description=original_route.description, - response_description=original_route.response_description, - responses=original_route.responses.copy() if original_route.responses else None, - deprecated=original_route.deprecated, - name=original_route.name, - methods=original_route.methods.copy() if original_route.methods else None, - operation_id=original_route.operation_id, - response_model_include=original_route.response_model_include, - response_model_exclude=original_route.response_model_exclude, - response_model_by_alias=original_route.response_model_by_alias, - response_model_exclude_unset=original_route.response_model_exclude_unset, - response_model_exclude_defaults=original_route.response_model_exclude_defaults, - response_model_exclude_none=original_route.response_model_exclude_none, - include_in_schema=original_route.include_in_schema, - response_class=original_route.response_class, - dependency_overrides_provider=original_route.dependency_overrides_provider, - callbacks=original_route.callbacks.copy() if original_route.callbacks else None, - openapi_extra=( - original_route.openapi_extra.copy() - if original_route.openapi_extra - else None - ), - generate_unique_id_function=original_route.generate_unique_id_function, - ) - _new_route.path_regex = re.compile(_new_regex_string) - return _new_route - - -def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]: - """ - Groups API routes by their HTTP method. This is important to ensure that the URLPath is matched - correctly, as different methods may share the same URLPath (e.g. GET, POST PUT, DELETE). - """ - _method_to_routes = defaultdict( - list - ) # Automatically initializes new lists for new keys - for _route in api_routes: - if ( - isinstance(_route, APIRoute) and _route.methods - ): # Ensure methods exist (should always be true for valid routes) - for method in _route.methods: - _method_to_routes[method].append(clone_and_adapt_route(_route)) - - return dict(_method_to_routes) - - class BRAGArgument: """ This class is used to represent a BRAG API input argument. It is uses FastAPI annotation to define the arguments @@ -126,9 +48,7 @@ class BRAGArgument: self.is_nested = False def __repr__(self): - return f"{self.module}.{self.cls}" + ( - f"[{self.nested}]" if self.is_nested else "" - ) + return f"{self.module}.{self.cls}" + (f"[{self.nested}]" if self.is_nested else "") def convert_to_argument(arg: BRAGArgument, value: str) -> Any: @@ -185,13 +105,11 @@ class CleverBragAmqpAdapter(CleverThisServiceAdapter): routes: List[APIRoute], auth_provider: core.base.providers.auth.AuthProvider, ): - super().__init__(amq_configuration, None) - self.routes = group_routes_by_method(api_routes=routes) + super().__init__(amq_configuration, routes, None) self.auth_provider = auth_provider _amq_service: AMQService = AMQService(amq_configuration, self) - thread = Thread(target=_amq_service.run) - thread.start() - # thread.join() + self.thread = Thread(target=_amq_service.run) + self.thread.start() async def get_current_user(self, token: str) -> User: """ @@ -200,190 +118,5 @@ class CleverBragAmqpAdapter(CleverThisServiceAdapter): :return: User object """ return await self.auth_provider.auth_wrapper( - HTTPAuthorizationCredentials(f"Bearer {token}") + HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) ) - - # ================================================================================================== - # ===================== C l e v e r S w a r m A P I m e t h o d s ============================ - # ================================================================================================== - - async def get(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the GET requests for the CleverBRAG API. - :param message: message from the AMQP that contains the GET request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - return await self.handle_no_body_payload(message, auth_user, "GET") - - async def put(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the PUT requests for the CleverBRAG API. - :param message: message from the AMQP that contains the PUT request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - return await self.handle_body_payload(message, auth_user, "PUT") - - async def post(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the POST requests for the CleverBRAG API. - :param message: message from the AMQP that contains the POST request, including body and headers - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - return await self.handle_body_payload(message, auth_user, "POST") - - async def head(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the HEAD requests for the CleverBRAG API. - """ - return await self.handle_no_body_payload(message, auth_user, "HEAD") - - async def delete(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the DELETE requests for the CleverBRAG API. - """ - return await self.handle_no_body_payload(message, auth_user, "DELETE") - - async def options(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the OPTIONS requests for the CleverBRAG API. - """ - return await self.handle_no_body_payload(message, auth_user, "OPTIONS") - - async def patch(self, message: AMQMessage, auth_user: User) -> AMQResponse: - """ - Handles the PATCH requests for the CleverBRAG API. - """ - return await self.handle_no_body_payload(message, auth_user, "PATCH") - - async def handle_no_body_payload( - self, message: AMQMessage, auth_user: User, method: str - ) -> AMQResponse: - """ - Handles the requests that do not have a body payload. - :param message: message from the AMQP that contains the request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - _routes = self.routes.get(method, []) - _path, _args = AMQDataParser.parse_get_url(message.path) - for _route in _routes: - _match = _route.path_regex.match(_path) - if _match: - _args.update(_match.groupdict()) - _args["authenticated_user"] = auth_user - # If the path matches, call the corresponding function - return await CleverThisServiceAdapter.call_cleverthis_api( - message, - _args, - api_method=_route.endpoint, - success_code=_route.status_code, - loop=self.loop, - ) - - return AMQResponseFactory.of( - message.id, - 404, - "text/plain", - str(f"Destination location [{message.path}] does not exists").encode( - "utf-8" - ), - message.trace_info, - ) - - async def handle_body_payload( - self, message: AMQMessage, auth_user: User, method: str - ) -> AMQResponse: - """ - Handles the requests that have a body payload, including Multipart data. - :param message: message from the AMQP that contains the request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :param method: HTTP method (POST, PUT, etc.) - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - _form_data = AMQDataParser.parse_request_body(message) - _routes = self.routes.get(method, []) - _path, _query_args = AMQDataParser.parse_get_url(message.path) - for _route in _routes: - _match = _route.path_regex.match(_path) - if _match: - # If the path matches, call the corresponding function - _form_data.update(_match.groupdict()) - _form_data.update(_query_args) - for _body_param in _route.dependant.body_params: - if _body_param.name in _form_data: - _file_data = _form_data[_body_param.name] - actual_filepath = message.extra_data[_body_param.name] - _form_data[_body_param.name] = ( - _body_param.type_.__name__ == "UploadFile" - and UploadFile( - file=pathlib.Path(actual_filepath).open("rb"), - size=_file_data["size"], - filename=_file_data["filename"], - headers=Headers( - {"Content-Type": _file_data["mediaType"]} - ), - ) - or _form_data[_body_param.name] - ) - - _verified_args = {} - _required_args = getattr(_route.endpoint, "__annotations__", {}) - for arg in _required_args: - if arg in _form_data: - _verified_args[arg] = _form_data[arg] - else: - # needed for /login input - if arg == "authenticated_user": - _verified_args["authenticated_user"] = auth_user - if _required_args[arg].__name__ == "OAuth2PasswordRequestForm": - try: - _verified_args[arg] = OAuth2PasswordRequestForm( - **_form_data - ) - except Exception as e: - print(f"Error: {e}") - - return await CleverThisServiceAdapter.call_cleverthis_api( - message, - _verified_args, - api_method=_route.endpoint, - success_code=_route.status_code if _route.status_code else 200, - loop=self.loop, - ) - - return AMQResponseFactory.of( - message.id, - 404, - "text/plain", - str(f"Destination location [{message.path}] does not exists").encode( - "utf-8" - ), - message.trace_info, - ) - - -if __name__ == "__main__": - - # Example usage - r2rapp = asyncio.run(create_r2r_app()) - - adapter: CleverBragAmqpAdapter = CleverBragAmqpAdapter( - AMQConfiguration(""), r2rapp.app.routes, r2rapp.providers.auth - ) - - for method, routes in adapter.routes.items(): - print(f"Method: {method}") - for _route in routes: - print( - f" Route: {_route.path} -> {_route.endpoint.__module__}.{_route.endpoint.__name__}" - ) - _args = getattr(_route.endpoint, "__annotations__", {}) - _verified_args = [] - for arg in _args: - _arg = BRAGArgument(_args[arg]) - _verified_args.append(_arg) - print(f" Arguments: {', '.join(map(str, _verified_args))}") - print("---------------------------") diff --git a/cleverbrag/main.py b/cleverbrag/main.py new file mode 100644 index 0000000..e8f97a6 --- /dev/null +++ b/cleverbrag/main.py @@ -0,0 +1,41 @@ +from clever_brag_adapter import BRAGArgument, CleverBragAmqpAdapter + +from amqp.config.amq_configuration import AMQConfiguration + + +class R2RApp: + pass + + +def create_r2r_app(): + return R2RApp() + + +if __name__ == "__main__": + + # + # r2rapp: R2RApp = await create_r2r_app() + # + # In the standalone test context, need to start the loop as well. In the actual BRAG context, + # use 'await' instead. + r2rapp: R2RApp = create_r2r_app() + + adapter: CleverBragAmqpAdapter = CleverBragAmqpAdapter( + AMQConfiguration("./application.properties.local"), r2rapp.app.routes, r2rapp.providers.auth + ) + + for method, routes in adapter.routes.items(): + print(f"Method: {method}") + for _route in routes: + print( + f" Route: {_route.path} -> {_route.endpoint.__module__}.{_route.endpoint.__name__}" + ) + _args = getattr(_route.endpoint, "__annotations__", {}) + _verified_args = [] + for arg in _args: + _arg = BRAGArgument(_args[arg]) + _verified_args.append(_arg) + print(f" Arguments: {', '.join(map(str, _verified_args))}") + print("---------------------------") + print("Press ^C to exit...") + adapter.thread.join() diff --git a/cleverswarm/clever_swarm_adapter.py b/cleverswarm/clever_swarm_adapter.py index 13b781e..d5cd8b1 100644 --- a/cleverswarm/clever_swarm_adapter.py +++ b/cleverswarm/clever_swarm_adapter.py @@ -5,129 +5,51 @@ It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the C """ import importlib -import pathlib -import re -from collections import defaultdict from threading import Thread -from typing import Optional, List, Dict -from aiohttp import ClientSession -from fastapi.routing import APIRoute -from fastapi.security import OAuth2PasswordRequestForm -from starlette.datastructures import UploadFile, Headers +from typing import Any, List, Optional # CleverSwarm specific imports import core.deps -from rest.v0 import api +from fastapi.routing import APIRoute from schemas.user_schema import UserSchema -from amqp.adapter.data_parser import AMQDataParser - -# AMQP specific imports -from amqp.adapter.data_response_factory import AMQResponseFactory -from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage +from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter from amqp.config.amq_configuration import AMQConfiguration -from amqp.model.model import AMQResponse, AMQErrorMessage from amqp.service.amq_service import AMQService - -# Suffix for the method name with serialization wrapper, prefer use this one over the orig endpoint PREFERRED_SUFFIX = "_json" -def clone_and_adapt_route(original_route: APIRoute) -> APIRoute: +# ================================================================================================= +# =========================== S u p p o r t i n g f u n c t i o n s ============================= +# ================================================================================================= +def get_callable_for_name(module_name: str, function_name: str) -> Optional[callable]: """ - Creates a clone of an APIRoute. It checks if there's preferred endpoint with json serialization. - It also adjusts the path regex to ensure it matches the path in different API versions. + Get the function/Callable for the module_name.function_name if it exists in the module, else return None. """ - - # Check if the route has a preferred endpoint with json serialization - _preferred_endpoint = get_prefered_endpoint( - getattr(original_route.endpoint, "__module__"), - original_route.endpoint.__name__ + PREFERRED_SUFFIX, - ) - # Need to adapt also the path regex to ensure it matches the path in different API versions - # Ensures the leading '^' is followed by any match and there's no '/' before final '$' - _new_regex_string = ( - "^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$" - ) - _new_route: APIRoute = APIRoute( - path=original_route.path, - endpoint=( - _preferred_endpoint if _preferred_endpoint else original_route.endpoint - ), - response_model=original_route.response_model, - status_code=original_route.status_code, - tags=original_route.tags.copy() if original_route.tags else None, - dependencies=( - original_route.dependencies.copy() if original_route.dependencies else None - ), - summary=original_route.summary, - description=original_route.description, - response_description=original_route.response_description, - responses=original_route.responses.copy() if original_route.responses else None, - deprecated=original_route.deprecated, - name=original_route.name, - methods=original_route.methods.copy() if original_route.methods else None, - operation_id=original_route.operation_id, - response_model_include=original_route.response_model_include, - response_model_exclude=original_route.response_model_exclude, - response_model_by_alias=original_route.response_model_by_alias, - response_model_exclude_unset=original_route.response_model_exclude_unset, - response_model_exclude_defaults=original_route.response_model_exclude_defaults, - response_model_exclude_none=original_route.response_model_exclude_none, - include_in_schema=original_route.include_in_schema, - response_class=original_route.response_class, - dependency_overrides_provider=original_route.dependency_overrides_provider, - callbacks=original_route.callbacks.copy() if original_route.callbacks else None, - openapi_extra=( - original_route.openapi_extra.copy() - if original_route.openapi_extra - else None - ), - generate_unique_id_function=original_route.generate_unique_id_function, - ) - _new_route.path_regex = re.compile(_new_regex_string) - return _new_route - - -def get_prefered_endpoint(module_name: str, function_name: str) -> Optional[callable]: - """Get the function if it exists in the module, else return None.""" try: _module = importlib.import_module(module_name) - _prefered_endpoint = getattr(_module, function_name) - return _prefered_endpoint if callable(_prefered_endpoint) else None + _preferred_endpoint = getattr(_module, function_name) + return _preferred_endpoint if callable(_preferred_endpoint) else None except Exception: return None -def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]: - """ - Groups API routes by their HTTP method. This is important to ensure that the URLPath is matched - correctly, as different methods may share the same URLPath (e.g. GET, POST PUT, DELETE). - """ - _method_to_routes = defaultdict( - list - ) # Automatically initializes new lists for new keys - for _route in api_routes: - if ( - _route.methods - ): # Ensure methods exist (should always be true for valid routes) - for method in _route.methods: - _method_to_routes[method].append(clone_and_adapt_route(_route)) - - return dict(_method_to_routes) - - +# ================================================================================================= +# ======================== C l e v e r S w a r m A M Q A d a p t e r ======================== +# ================================================================================================= class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): """ - CleverSwarm AMQP Adapter for CleverSwarm API. Implements the interface to CleverSwarm API. + CleverSwarm AMQP Adapter for CleverSwarm API. Implements the CleverSwarm specific logic that overrides + common logic defined by AMQ Adapter library. The specific logic here is: + + 1. Extract subject from JWT token and convert it to UserSchema to avail authorized user argument + 2. Find JSON wrapper for endpoint Callable, and if exists, use it instead of the original endpoint. + """ - def __init__( - self, amq_configuration: AMQConfiguration, session: ClientSession = None - ): - super().__init__(amq_configuration, session) - self.routes = group_routes_by_method(api_routes=api.api_router.routes) + def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]): + super().__init__(amq_configuration, routes=routes, session=None) _amq_service: AMQService = AMQService(amq_configuration, self) self.thread = Thread(target=_amq_service.run) self.thread.start() @@ -135,170 +57,25 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): async def get_current_user(self, token: str) -> UserSchema: """ Overrides the default get_current_user method to use the token from the AMQP message. + In CleverSwarm context, it means simply to call provided get_current_user function. + :param token: JWT token from the AMQP message + :return: UserSchema object """ return await core.deps.get_current_user(token) - # ================================================================================================== - # ===================== C l e v e r S w a r m A P I m e t h o d s ============================ - # ================================================================================================== + def get_endpoint(self, endpoint: Any) -> Any: + """ + If the specific endpoint has a matching JSON wrapper (identified as {endpoint.name}_json ), + then use it as preferred way to invoke the endpoint. - async def get(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - """ - Handles the GET requests for the CleverSwarm API. - :param message: message from the AMQP that contains the GET request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - return await self.handle_no_body_payload(message, auth_user, "GET") + :param endpoint: the Callable for the endpoint - async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: + :return: new Callable representing the JSON wrapper endpoint, if it exists or the original Callable. """ - Handles the PUT requests for the CleverSwarm API. - :param message: message from the AMQP that contains the PUT request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - return await self.handle_body_payload(message, auth_user, "PUT") - - async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - """ - Handles the POST requests for the CleverSwarm API. - :param message: message from the AMQP that contains the POST request, including body and headers - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - return await self.handle_body_payload(message, auth_user, "POST") - - async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - """ - Handles the HEAD requests for the CleverSwarm API. - """ - return await self.handle_no_body_payload(message, auth_user, "HEAD") - - async def delete(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - """ - Handles the DELETE requests for the CleverSwarm API. - """ - return await self.handle_no_body_payload(message, auth_user, "DELETE") - - async def options(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - """ - Handles the OPTIONS requests for the CleverSwarm API. - """ - return await self.handle_no_body_payload(message, auth_user, "OPTIONS") - - async def patch(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - """ - Handles the PATCH requests for the CleverSwarm API. - """ - return await self.handle_no_body_payload(message, auth_user, "PATCH") - - async def handle_no_body_payload( - self, message: AMQMessage, auth_user: UserSchema, method: str - ) -> AMQResponse: - """ - Handles the requests that do not have a body payload. - :param message: message from the AMQP that contains the request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - _routes = self.routes.get(method, []) - _path, _args = AMQDataParser.parse_get_url(message.path) - for _route in _routes: - _match = _route.path_regex.match(_path) - if _match: - _args.update(_match.groupdict()) - _args["authenticated_user"] = auth_user - # If the path matches, call the corresponding function - return await CleverThisServiceAdapter.call_cleverthis_api( - message, - _args, - api_method=_route.endpoint, - success_code=_route.status_code, - loop=self.loop, - ) - - return AMQResponseFactory.of_error_message( - message.id, - AMQErrorMessage( - 404, - "Not Found", - f"Destination location [{message.path}] does not exists", - ), - message.trace_info, - ) - - async def handle_body_payload( - self, message: AMQMessage, auth_user: UserSchema, method: str - ) -> AMQResponse: - """ - Handles the requests that have a body payload, including Multipart data. - :param message: message from the AMQP that contains the request - :param auth_user: user authenticated by the API Gateway, as UserSchem - :param method: HTTP method (POST, PUT, etc.) - :return: AMQResponse to be sent back to the API Gateway via AMQP - """ - _form_data = AMQDataParser.parse_request_body(message) - _routes = self.routes.get(method, []) - _path, _query_args = AMQDataParser.parse_get_url(message.path) - for _route in _routes: - _match = _route.path_regex.match(_path) - if _match: - # If the path matches, call the corresponding function - _form_data.update(_match.groupdict()) - _form_data.update(_query_args) - for _body_param in _route.dependant.body_params: - if _body_param.name in _form_data: - _file_data = _form_data[_body_param.name] - _form_data[_body_param.name] = ( - _body_param.type_.__name__ == "UploadFile" - and message.extra_data.get(_body_param.name) - and UploadFile( - file=pathlib.Path( - message.extra_data.get(_body_param.name) - ).open("rb"), - size=_file_data["size"], - filename=_file_data["filename"], - headers=Headers( - {"Content-Type": _file_data["mediaType"]} - ), - ) - or _form_data[_body_param.name] - ) - - _verified_args = {} - _required_args = getattr(_route.endpoint, "__annotations__", {}) - for arg in _required_args: - if arg in _form_data: - _verified_args[arg] = _form_data[arg] - else: - # needed for /login input - if arg == "authenticated_user": - _verified_args["authenticated_user"] = auth_user - if _required_args[arg].__name__ == "OAuth2PasswordRequestForm": - try: - _verified_args[arg] = OAuth2PasswordRequestForm( - **_form_data - ) - except Exception as e: - print(f"Error: {e}") - - return await CleverThisServiceAdapter.call_cleverthis_api( - message, - _verified_args, - api_method=_route.endpoint, - success_code=_route.status_code if _route.status_code else 200, - loop=self.loop, - ) - - return AMQResponseFactory.of_error_message( - message.id, - AMQErrorMessage( - 404, - "Not Found", - f"Destination location [{message.path}] does not exists", - ), - message.trace_info, + _preferred_endpoint = get_callable_for_name( + getattr(endpoint, "__module__"), + endpoint.__name__ + PREFERRED_SUFFIX, ) + return _preferred_endpoint if _preferred_endpoint is not None else endpoint diff --git a/cleverswarm/main.py b/cleverswarm/main.py index ca7338c..9c55f02 100644 --- a/cleverswarm/main.py +++ b/cleverswarm/main.py @@ -1,7 +1,7 @@ -from amqp.config.amq_configuration import AMQConfiguration - from clever_swarm_adapter import CleverSwarmAmqpAdapter +from amqp.config.amq_configuration import AMQConfiguration + if __name__ == "__main__": adapter = CleverSwarmAmqpAdapter(AMQConfiguration("./application.properties.local")) print("Press ^C to exit...") diff --git a/pyproject.toml b/pyproject.toml index 01a71e0..c86f5b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "amq_adapter" -version = "0.2.30" +version = "0.2.31" description = "The Python implementation of the AMQ Adapter for CleverMicro" readme = "README.md" authors = [ @@ -36,11 +36,18 @@ dependencies = [ where = ["."] include = ["amqp*"] +[tool.black] +line-length = 100 + [project.optional-dependencies] dev = [ "pytest>=7.0", "pytest-cov>=4.0", + "pytest-asyncio", "build", - "twine" + "twine", + "black", + "isort", + "flake8", + "pre-commit" ] - diff --git a/tests/adapter/test_data_response_factory.py b/tests/adapter/test_data_response_factory.py index 68f758e..85f31a8 100644 --- a/tests/adapter/test_data_response_factory.py +++ b/tests/adapter/test_data_response_factory.py @@ -2,6 +2,7 @@ from unittest import TestCase from amqp.adapter.data_message_factory import DataMessageFactory from amqp.adapter.data_response_factory import DataResponseFactory +from amqp.model.model import DataMessageMagicByte from amqp.model.snowflake_id import SnowflakeId @@ -12,7 +13,7 @@ class TestDataResponseFactory(TestCase): _g = DataMessageFactory(111) _resp = _f.create_async_response_message(_g.generate_snowflake_id()) self.assertEqual(200, _resp.response_code) - self.assertEqual(b"OK", _resp.response) + self.assertEqual(b"OK", _resp.base64body()) def test_serialize(self): _f = DataResponseFactory() @@ -22,7 +23,7 @@ class TestDataResponseFactory(TestCase): _ser = _f.serialize(_resp) self.assertEqual( 2, - str(_ser[2:]).index(_f.MAGIC_BYTE_HTTP_REQUEST + _snowflakeId.as_string()), + str(_ser[2:]).index(DataMessageMagicByte.HTTP_REQUEST.value + _snowflakeId.as_string()), "Magic byte and SnowflakeId not found", ) self.assertGreater(str(_ser[2:]).index("|e=|"), 2) @@ -37,4 +38,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.base64body())