Modify CleverSwarm adapter to use reflection to match URLPath to endpoint

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,
+        success_code: int
-        authenticated_user: Any
 ) -> 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 api_method(**args)
-        result: str = await coroutine_call
         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)

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 collections import defaultdict
-from aiohttp import ClientSession, ClientResponse
+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<job_id>[^\?]+)\?index=(?P<index>\d+)"
+# Suffix for the method name with serialization wrapper, prefer use this one over the orig endpoint
-GET_JOB_STATUS = r".*/jobs/(?P<job_id>.*)"
+PREFERRED_SUFFIX = '_json'
-GET_KG = r".*/unstructured/(?P<job_id>.*)"
+
-PUT_BENCHMARK_KG = r".*/benchmark/(?P<job_id>.*)"
+def clone_and_adapt_route(original_route: APIRoute) -> APIRoute:
-DUMMY = r"(?P<job_id>.*)"
+    """
+    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
 
 
-#
+def get_prefered_endpoint(module_name: str, function_name: str) -> Optional[callable]:
-# Following methods are wrappers to actual CleverSwarm endpoints, so I can refer to the CleverSwarm method and
+    """Get the function if it exists in the module, else return None."""
-# pass it as input parameter using general Callable[[Any, UserSchema], str] type
+    try:
-#
+        module = importlib.import_module(module_name)
-async def cleverswarm_create_benchmark_job(args: Any, authenticated_user: UserSchema) -> str:
+        prefered_endpoint = getattr(module, function_name)
-    return str(await benchmark.create_benchmark_job(**args, authenticated_user=authenticated_user))
+        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:
+def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]:
-    return str(await unstructured.create_extract_with_ontology(**args, authenticated_user=authenticated_user))
+    """
+    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)
-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))
 
 
 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)
+        self.routes = group_routes_by_method(api_routes=api.api_router.routes)
-        amq_service.rabbit_mq_consumer.channel.start_consuming()
+        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
+        Overrides the default get_current_user method to use the token from the AMQP message.
-            payload = {
+        :param token: JWT token from the AMQP message
-                "sub": "0",  # Standard claim for subject (user)
+        :return: UserSchema object
-                "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)
-
         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 self.handle_no_body_payload(message, auth_user, 'GET')
-            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)
 
     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)
+        return await self.handle_body_payload(message, auth_user, 'PUT')
-        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)
 
     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)
+        return await self.handle_body_payload(message, auth_user, 'POST')
-        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()
 
     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(
+        Handles the DELETE requests for the CleverSwarm API.
-                message,
+        """
-                GET_JOB_STATUS,
+        return await self.handle_no_body_payload(message, auth_user, 'DELETE')
-                cleverswarm_retry_job,
-                authenticated_user=auth_user
-            )
 
-        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)