Stanislav Hejny
82 lines
3.6 KiB
Python
82 lines
3.6 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
|
|
from threading import Thread
|
|
from typing import Any, List, Optional
|
|
|
|
# CleverSwarm specific imports
|
|
import core.deps
|
|
from fastapi.routing import APIRoute
|
|
from schemas.user_schema import UserSchema
|
|
|
|
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
from amqp.service.amq_service import AMQService
|
|
|
|
PREFERRED_SUFFIX = "_json"
|
|
|
|
|
|
# =================================================================================================
|
|
# =========================== 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]:
|
|
"""
|
|
Get the function/Callable for the module_name.function_name if it exists in the module, else return None.
|
|
"""
|
|
try:
|
|
_module = importlib.import_module(module_name)
|
|
_preferred_endpoint = getattr(_module, function_name)
|
|
return _preferred_endpoint if callable(_preferred_endpoint) else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# =================================================================================================
|
|
# ======================== 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 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, 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()
|
|
|
|
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)
|
|
|
|
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.
|
|
|
|
:param endpoint: the Callable for the endpoint
|
|
|
|
:return: new Callable representing the JSON wrapper endpoint, if it exists or the original Callable.
|
|
"""
|
|
_preferred_endpoint = get_callable_for_name(
|
|
getattr(endpoint, "__module__"),
|
|
endpoint.__name__ + PREFERRED_SUFFIX,
|
|
)
|
|
return _preferred_endpoint if _preferred_endpoint is not None else endpoint
|