clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat/#1 - refactor code to allow make it as library and to separate the AMQ code from dependencies needed to write the CleverSwarm interface

Browse Source
This commit is contained in:
Stanislav Hejny
2025-02-26 10:51:17 +00:00
parent eb26eec767
commit 1f9040729f
20 changed files with 294 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/adapter/clever_swarm_adapter.py → amqp/adapter/cleverthis_service_adapter.py
+42 -150
View File
@@ -11,11 +11,8 @@ from amqp.adapter.amq_response_factory import AMQResponseFactory
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 core.deps import get_current_user
from schemas.user_schema import UserSchema
from cleverswarm.core.deps import get_current_user
from cleverswarm.schemas.user_schema import UserSchema
import json
from urllib.parse import parse_qs
@@ -94,17 +91,6 @@ def parse_request_body(message: AMQMessage):
else:
raise ValueError(f"Unsupported Content-Type: {content_type}")
BENCHMARK = '/benchmark'
JOBS = '/jobs/'
UNSTRUCTURED = '/unstructured'
GET_BENCHMARK_KG = r".*/benchmark/(?P<request_id>[^\?]+)\?index=(?P<index>\d+)"
GET_JOB_STATUS = r".*/jobs/(?P<request_id>.*)"
GET_KG = r".*/unstructured/(?P<request_id>.*)"
DUMMY = r"(?P<request_id>.*)"
# ============= POST Helper function =============
async def call_cleverswarm_post_put_api(
@@ -128,33 +114,6 @@ async def call_cleverswarm_post_put_api(
return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
#
# 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))
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))
async def cleverswarm_create_extract_with_wildcards(args: Any, authenticated_user: UserSchema) -> str:
return str(await unstructured.create_extract_with_ontology(**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))
# ============= GET Helper function =============
async def call_cleverswarm_get_api(
@@ -179,31 +138,32 @@ async def call_cleverswarm_get_api(
return AMQResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
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_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 CleverSwarmAdapter:
# ============= CleverThisServiceAdapter =============
class CleverThisServiceAdapter:
"""
A base class for CleverThis service adapter that handles the incoming AMQP messages and routes them to the
appropriate CleverThisService API endpoint.
"""
def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None):
self.amq_configuration = amq_configuration
self.session = session
self.service_host = self.amq_configuration.dispatch.service_host
self.service_port = self.amq_configuration.dispatch.service_port
async def get_current_user(self, token: str):
"""
To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
in format appropriate for the service.
:return: A user object or None if the user is not authenticated.
"""
return None
async def on_message(self, message) -> AMQResponse:
"""
Called when a message is received from the AMQP.
:param message:
:return: AMQP response class with operation status code and optional error details.
"""
DEBUG_NO_AUTH = 1
_auth: str = message.headers.get('Authorization', '')
_auth_user: UserSchema | None = None
@@ -231,47 +191,17 @@ class CleverSwarmAdapter:
return AMQResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
# ==================================================================================================
# ===================== C l e v e r S w a r m A P I m e t h o d s ============================
# ===================== C l e v e r T h i s 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.
Handles the GET requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
: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
"""
if BENCHMARK in message.path:
return await call_cleverswarm_get_api(
message,
GET_BENCHMARK_KG,
cleverswarm_get_benchmark_kg,
authenticated_user=auth_user
)
if JOBS in message.path:
return await call_cleverswarm_get_api(
message,
GET_JOB_STATUS,
cleverswarm_get_job_status,
authenticated_user=auth_user
)
if message.path.endswith('/jobs'):
return await call_cleverswarm_get_api(
message,
DUMMY,
cleverswarm_jobs_get_requests,
authenticated_user=auth_user
)
if UNSTRUCTURED in message.path:
return await call_cleverswarm_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}"
@@ -287,29 +217,12 @@ class CleverSwarmAdapter:
async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
"""
Handles the PUT requests for the CleverSwarm API.
Handles the PUT requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
: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
"""
parser: CleverMultiPartParser = CleverMultiPartParser(message)
if BENCHMARK in message.path:
return await call_cleverswarm_post_put_api(
message,
parser.form_data,
cleverswarm_create_benchmark_job,
201,
authenticated_user=auth_user
)
if JOBS in message.path:
return await call_cleverswarm_get_api(
message,
GET_JOB_STATUS,
cleverswarm_retry_job,
authenticated_user=auth_user
)
return await self.handle_possible_form(
self.session.put(
f"http://{self.service_host}:{self.service_port}{message.path}",
@@ -321,38 +234,12 @@ class CleverSwarmAdapter:
async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
"""
Handles the POST requests for the CleverSwarm API.
Handles the POST requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
: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
"""
parser: CleverMultiPartParser = CleverMultiPartParser(message)
if BENCHMARK in message.path:
return await call_cleverswarm_post_put_api(
message,
parser.form_data,
cleverswarm_create_benchmark_job,
201,
authenticated_user=auth_user
)
if UNSTRUCTURED in message.path:
if 'with_ontology' in message.path:
return await call_cleverswarm_post_put_api(
message,
parser.form_data,
cleverswarm_create_extract_with_ontology,
200,
authenticated_user=auth_user
)
if 'with_wildcards' in message.path:
return await call_cleverswarm_post_put_api(
message,
parser.form_data,
cleverswarm_create_extract_with_wildcards,
200,
authenticated_user=auth_user
)
return await self.handle_possible_form(
self.session.post(
f"http://{self.service_host}:{self.service_port}{message.path}",
@@ -366,18 +253,23 @@ class CleverSwarmAdapter:
return await request_coroutine
async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
"""
Handles the HEAD requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
: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
"""
raise NotImplementedError
async def delete(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
parser: CleverMultiPartParser = CleverMultiPartParser(message)
if JOBS in message.path:
return await call_cleverswarm_get_api(
message,
GET_JOB_STATUS,
cleverswarm_retry_job,
authenticated_user=auth_user
)
"""
Handles the DELETE requests for the CleverThis API. to be overriden in subclass. This is a base implementation.
It will attempt to call the REST endpoint directly using HTTP protocol.
: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
"""
raise NotImplementedError
def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
amqp/service/amq_service.py
+3 -4
View File
@@ -1,4 +1,3 @@
import asyncio
import logging
import logging.config
import os
@@ -7,11 +6,11 @@ from asyncio import Future
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_route_factory import AMQRouteFactory
from amqp.adapter.clever_swarm_adapter import CleverSwarmAdapter
from cleverswarm.clever_swarm_adapter import CleverSwarmAdapter
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage, AMQRoute
from amqp.model.model import AMQMessage
from amqp.rabbitmq.rabbit_mq_consumer import RabbitMQConsumer
from amqp.service.service_message_handler import AMQServiceMessageHandler
from amqp.service.tracing import initialize_trace
@@ -91,4 +90,4 @@ if __name__ == "__main__":
amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAdapter(amq_configuration))
amq_service.rabbit_mq_consumer.channel.start_consuming()
# amq_service.consumer_thread.cancel()
# amq_service.rabbit_mq_consumer.connection.close()
# amq_service.rabbit_mq_consumer.connection.close()
amqp/service/service_message_handler.py
+1 -1
View File
@@ -13,7 +13,7 @@ from pika.channel import Channel
from amqp.adapter.amq_message_factory import AMQMessageFactory
from amqp.adapter.amq_response_factory import AMQResponseFactory
from amqp.adapter.clever_swarm_adapter import CleverSwarmAdapter
from cleverswarm.clever_swarm_adapter import CleverSwarmAdapter
from amqp.model.model import AMQResponse, AMQMessage
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
from amqp.router.serializer import map_as_string