diff --git a/amqp/adapter/cleverthis_service_adapter.py b/amqp/adapter/cleverthis_service_adapter.py index 7baa410..88e0755 100644 --- a/amqp/adapter/cleverthis_service_adapter.py +++ b/amqp/adapter/cleverthis_service_adapter.py @@ -8,8 +8,9 @@ import logging import re import sys import traceback +from re import Pattern from typing import Dict, List, Tuple, Callable, Coroutine, AnyStr, Any -from urllib.parse import parse_qs +from urllib.parse import parse_qs, urlparse from xml.etree import ElementTree from aiohttp import ClientSession, ClientResponse @@ -84,8 +85,7 @@ async def call_cleverthis_post_put_api( message: DataMessage, args: Any, api_method: Callable[[Any, Any], Coroutine], - success_code: int, - authenticated_user: Any + success_code: int ) -> DataResponse: """ Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary. @@ -95,12 +95,10 @@ async def call_cleverthis_post_put_api( :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 - :param authenticated_user: Authenticated user. The actual class depends on the service :return: DataResponse class """ try: - coroutine_call: Coroutine = api_method(args, authenticated_user) - result: str = await coroutine_call + result: str = await api_method(**args) return DataResponseFactory.of( message.id, success_code, message.headers.get("Content-Type", "application/json"), result.encode('utf-8'), message.trace_info @@ -116,10 +114,27 @@ async def call_cleverthis_post_put_api( return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info) +def parse_get_url(url: str) -> Tuple[str, Dict[str, str]]: + """ + Parses an HTTP GET URL and returns (context_path, query_params_dict) + + Args: + url: The URL string to parse (e.g., "http://example.com/path?foo=1&bar=2&foo=3") + + Returns: + tuple: (context_path, query_params_dict) + - context_path: The path part of the URL (e.g., "/path") + - query_params_dict: Query parameters where values are str for single value or list of str for multivalue + """ + 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()} + return parsed.path, simplified_params async def call_cleverthis_get_api( message: DataMessage, - pattern_or_data: str | dict, + pattern_or_data: Pattern| str | dict, api_method: Callable[[Tuple[AnyStr | Any, ...], Any], Coroutine], authenticated_user: Any ) -> DataResponse: @@ -129,7 +144,7 @@ async def call_cleverthis_get_api( invokes provided Callable that is expected to be a GET API method, and returns the response as DataResponse :param message: DataMessage containing the details of the call :param pattern_or_data: If there are path variables, use this pattern to extract them - :param api_method: Callable - the method to invoke. Last parameter is the authenticated user + :param api_method: Callable - the method to invoke. :param authenticated_user: Authenticated user. The actual class depends on the service :return: """ @@ -137,6 +152,14 @@ async def call_cleverthis_get_api( args = None if isinstance(pattern_or_data, dict): args = pattern_or_data + elif isinstance(pattern_or_data, Pattern): + path, args = pydantic_serializer.parse_url_query(message.path) + match = pattern_or_data.match(path) + if match: + args.update(match.groupdict()) + else: + _msg = f"Unexpected URL: {message.path}, expected pattern: {pattern_or_data}" + return DataResponseFactory.of(message.id, 404, "text/plain", _msg.encode('utf-8'), message.trace_info) else: path, args = pydantic_serializer.parse_url_query(message.path) match = re.search(pattern_or_data, path) diff --git a/cleverswarm/clever_swarm_adapter.py b/cleverswarm/clever_swarm_adapter.py index 15a4728..333501d 100644 --- a/cleverswarm/clever_swarm_adapter.py +++ b/cleverswarm/clever_swarm_adapter.py @@ -1,108 +1,124 @@ -import logging +""" +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 typing import Tuple, AnyStr, Any -from aiohttp import ClientSession, ClientResponse +from collections import defaultdict +from typing import Optional, List, Dict +from aiohttp import ClientSession +from fastapi.routing import APIRoute from starlette.datastructures import UploadFile, Headers -import jwt - -from datetime import datetime, timedelta +# CleverSwarm specific imports import core.deps -from amqp.adapter.amq_response_factory import AMQResponseFactory +from rest.v0 import api +from schemas.user_schema import UserSchema + +# AMQP specific imports +from amqp.adapter.data_response_factory import AMQResponseFactory from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, call_cleverthis_get_api, \ - call_cleverthis_post_put_api, parse_request_body + call_cleverthis_post_put_api, parse_get_url from amqp.service.multipart_parser import CleverMultiPartParser from amqp.config.amq_configuration import AMQConfiguration from amqp.model.model import AMQMessage, AMQResponse - -import rest.v0.endpoints.benchmark as benchmark -import rest.v0.endpoints.jobs as jobs -import rest.v0.endpoints.unstructured as unstructured from amqp.service.amq_service import AMQService -from core.configs import settings -from schemas.user_schema import UserSchema -BENCHMARK = '/benchmark' -JOBS = '/jobs/' -JOB_DETAILS = '/jobs/details/' -UNSTRUCTURED = '/unstructured' -GET_BENCHMARK_KG = r".*/benchmark/(?P[^\?]+)\?index=(?P\d+)" -GET_JOB_STATUS = r".*/jobs/(?P.*)" -GET_KG = r".*/unstructured/(?P.*)" -PUT_BENCHMARK_KG = r".*/benchmark/(?P.*)" -DUMMY = r"(?P.*)" +# 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 + _new_regex_string = '^.*' + original_route.path_regex.pattern.lstrip('^') # Ensures '^' is only at the start + _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 -# -# Following methods are wrappers to actual CleverSwarm endpoints, so I can refer to the CleverSwarm method and -# pass it as input parameter using general Callable[[Any, UserSchema], str] type -# -async def cleverswarm_create_benchmark_job(args: Any, authenticated_user: UserSchema) -> str: - return str(await benchmark.create_benchmark_job(**args, authenticated_user=authenticated_user)) +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 -async def cleverswarm_create_extract_with_ontology(args: Any, authenticated_user: UserSchema) -> str: - return str(await unstructured.create_extract_with_ontology(**args, authenticated_user=authenticated_user)) +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)) - -async def cleverswarm_create_extract_with_wildcards(args: Any, authenticated_user: UserSchema) -> str: - return str(await unstructured.create_extract_with_wildcards(**args, authenticated_user=authenticated_user)) - - -async def cleverswarm_update_benchmark_job(args: Any, authenticated_user: UserSchema) -> str: - return str(await benchmark.update_benchmark_job(**args, authenticated_user=authenticated_user)) - - -async def cleverswarm_retry_job(args: Any, authenticated_user: UserSchema) -> str: - return str(await jobs.retry_job(args[0], authenticated_user=authenticated_user)) - - -async def cleverswarm_delete_job(args: Any, authenticated_user: UserSchema) -> str: - return str(await jobs.delete_job(args[0], authenticated_user=authenticated_user)) - - -async def cleverswarm_get_benchmark_kg(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: - return str(await benchmark.get_benchmark_kg(args[0], args[1], authenticated_user=authenticated_user)) - - -async def cleverswarm_get_job_status(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: - return str(await jobs.get_job_status(args[0], authenticated_user=authenticated_user)) - -async def cleverswarm_get_job_details(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: - return str(await jobs.get_job_details(args[0], authenticated_user=authenticated_user)) - - -async def cleverswarm_jobs_get_requests(args: Tuple[AnyStr | Any, ...], - authenticated_user: UserSchema) -> str: - return str(await jobs.get_requests(authenticated_user=authenticated_user)) - - -async def cleverswarm_get_kg(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str: - return str(await unstructured.get_kg(args[0], authenticated_user=authenticated_user)) + 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) - amq_service: AMQService = AMQService(amq_configuration,self) - amq_service.rabbit_mq_consumer.channel.start_consuming() + self.routes = group_routes_by_method(api_routes=api.api_router.routes) + amq_service: AMQService = AMQService(amq_configuration, self) + amq_service.rabbit_mq_client.channel.start_consuming() async def get_current_user(self, token: str) -> UserSchema: - if token == 'DEBUG_NO_AUTH': - # Payload with username and expiration - payload = { - "sub": "0", # Standard claim for subject (user) - "username": "user", # Custom claim (optional) - "iat": datetime.utcnow(), # Issued at time - "exp": datetime.utcnow() + timedelta(days=365), # Expires in 1 year - } - # Generate the JWT token - token = jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.ALGORITHM) - + """ + 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 ============================ # ================================================================================================== @@ -114,57 +130,7 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): :param auth_user: user authenticated by the API Gateway, as UserSchem :return: AMQResponse to be sent back to the API Gateway via AMQP """ - if BENCHMARK in message.path: - return await call_cleverthis_get_api( - message, - GET_BENCHMARK_KG, - cleverswarm_get_benchmark_kg, - authenticated_user=auth_user - ) - - if JOBS in message.path: - if JOB_DETAILS in message.path: - return await call_cleverthis_get_api( - message, - DUMMY, - cleverswarm_get_job_status, - authenticated_user=auth_user - ) - return await call_cleverthis_get_api( - message, - GET_JOB_STATUS, - cleverswarm_get_job_status, - authenticated_user=auth_user - ) - - if message.path.endswith('/jobs'): - return await call_cleverthis_get_api( - message, - DUMMY, - cleverswarm_jobs_get_requests, - authenticated_user=auth_user - ) - - if UNSTRUCTURED in message.path: - return await call_cleverthis_get_api( - message, - GET_KG, - cleverswarm_get_kg, - authenticated_user=auth_user - ) - - # 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 AMQResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type, - await _http_resp.read(), - message.trace_info) - return AMQResponseFactory.of(message.id, 404, "text/plain", - str(f"Destination location [{url}] does not exists").encode('utf-8'), - message.trace_info) + return await self.handle_no_body_payload(message, auth_user, 'GET') async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: """ @@ -173,37 +139,7 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): :param auth_user: user authenticated by the API Gateway, as UserSchem :return: AMQResponse to be sent back to the API Gateway via AMQP """ - parser: CleverMultiPartParser = CleverMultiPartParser(message) - if BENCHMARK in message.path: - match = re.search(PUT_BENCHMARK_KG, message.path) - if match: - parser.form_data['job_id'] = match.group('job_id') - return await call_cleverthis_post_put_api( - message, - parser.form_data, - cleverswarm_update_benchmark_job, - 201, - authenticated_user=auth_user - ) - - if JOBS in message.path: - return await call_cleverthis_get_api( - message, - GET_JOB_STATUS, - cleverswarm_retry_job, - authenticated_user=auth_user - ) - url = f"http://{self.service_host}:{self.service_port}{message.path}" - return await self.handle_possible_form( - self.session.put( - url, - data=message.body, - headers={**message.headers, **message.trace_info}, - ), - auth_user - ) if self.session else AMQResponseFactory.of(message.id, 404, "text/plain", - str(f"Destination location [{url}] does not exists").encode('utf-8'), - message.trace_info) + return await self.handle_body_payload(message, auth_user, 'PUT') async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: """ @@ -212,76 +148,96 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter): :param auth_user: user authenticated by the API Gateway, as UserSchem :return: AMQResponse to be sent back to the API Gateway via AMQP """ - _args = parse_request_body(message) - args = {} - if isinstance(_args, dict): - for key, file in _args['files'].items(): - if len(file) > 0: - filepath = message.extra_data[key] - ufile = UploadFile( - file=pathlib.Path(filepath).open("rb"), - size=file[0]['size'], - filename=file[0]['filename'], - headers=Headers({"Content-Type": file[0]['mediaType']}) - ) - args[key] = ufile - for key, obj in _args['form'].items(): - if key not in args: - args[key] = obj[0] if isinstance(obj, list) else obj - - if BENCHMARK in message.path: - return await call_cleverthis_post_put_api( - message, - args, - cleverswarm_create_benchmark_job, - 201, - authenticated_user=auth_user - ) - if UNSTRUCTURED in message.path: - if 'with_ontology' in message.path: - return await call_cleverthis_post_put_api( - message, - args, - cleverswarm_create_extract_with_ontology, - 200, - authenticated_user=auth_user - ) - if 'with_wildcards' in message.path: - return await call_cleverthis_post_put_api( - message, - args, - cleverswarm_create_extract_with_wildcards, - 200, - authenticated_user=auth_user - ) - url = f"http://{self.service_host}:{self.service_port}{message.path}" - return await self.handle_possible_form( - self.session.post( - url, - data=message.body, - headers={**message.headers, **message.trace_info}, - ), - auth_user - ) if self.session else AMQResponseFactory.of(message.id, 404, "text/plain", - str(f"Destination location [{url}] does not exists").encode('utf-8'), - message.trace_info) - - async def handle_possible_form(self, - message: AMQMessage, - request_coroutine, - auth_user: UserSchema) -> AMQResponse: - return await request_coroutine() + return await self.handle_body_payload(message, auth_user, 'POST') async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse: - raise NotImplementedError + """ + 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: - if JOBS in message.path: - return await call_cleverthis_get_api( - message, - GET_JOB_STATUS, - cleverswarm_retry_job, - authenticated_user=auth_user - ) + """ + Handles the DELETE requests for the CleverSwarm API. + """ + return await self.handle_no_body_payload(message, auth_user, 'DELETE') - raise NotImplementedError + 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 = 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 call_cleverthis_post_put_api( + message, + args, + api_method=route.endpoint, + success_code=route.status_code + ) + + 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 + """ + parser: CleverMultiPartParser = CleverMultiPartParser(message) + routes = self.routes.get(method, []) + path, query_args = 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 + parser.form_data.update(match.groupdict()) + parser.form_data.update(query_args) + parser.form_data['authenticated_user']=auth_user + for body_param in route.dependant.body_params: + if body_param.name in parser.form_data: + form_data = parser.form_data[body_param.name] + actual_filepath = message.extra_data[body_param.name] + parser.form_data[body_param.name] = body_param.type_.__name__ == 'UploadFile' and \ + UploadFile( + file=pathlib.Path(actual_filepath).open("rb"), + size=form_data['size'], + filename=form_data['filename'], + headers=Headers({"Content-Type": form_data['mediaType']}) + ) or parser.form_data[body_param.name] + + + return await call_cleverthis_post_put_api( + message, + parser.form_data, + api_method=route.endpoint, + success_code=route.status_code + ) + + return AMQResponseFactory.of(message.id, 404, "text/plain", + str(f"Destination location [{message.path}] does not exists").encode('utf-8'), + message.trace_info)