""" This module provides the CleverSwarm AMQP Adapter for handling API requests via AMQP. The code IS MEANT to be part of CleverSwarm codebase, and has direct dependencies on the CleverSwarm codebase. It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the CleverSwarm endpoints context. """ 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 # CleverSwarm specific imports import core.deps from rest.v0 import api 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.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQResponse 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: """ 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. """ # 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 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) class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): """ CleverSwarm AMQP Adapter for CleverSwarm API. Implements the interface to CleverSwarm API. """ 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) _amq_service: AMQService = AMQService(amq_configuration, self) self.thread = Thread(target=_amq_service.run) self.thread.start() async def get_current_user(self, token: str) -> UserSchema: """ Overrides the default get_current_user method to use the token from the AMQP message. :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 ============================ # ================================================================================================== 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") async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: """ 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( 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: 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( message.id, 404, "text/plain", str(f"Destination location [{message.path}] does not exists").encode( "utf-8" ), message.trace_info, )