Stanislav Hejny
303 lines
13 KiB
Python
303 lines
13 KiB
Python
"""
|
|
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]
|
|
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,
|
|
)
|