clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity
Files
a8d82b1e3d37655d2546ebdb19a4801e6ab3e41a
amq-adapter-python/amqp/adapter/cleverthis_service_adapter.py
T
Stanislav Hejny a8d82b1e3d #feat/1 - code cleanup for final release, part 1
2025-04-17 13:20:56 +01:00

418 lines
20 KiB
Python
Raw Blame History

"""
The layer that interfaces with the CleverThis service that this Adapter is integrating with.
Methods here will invoke the actual CleverThis Service code, which are passed as Callable object.
"""
import asyncio
import inspect
import json
import logging
import sys
import traceback
from asyncio import AbstractEventLoop
from typing import Dict, List, Tuple, Callable, Coroutine, Any
from urllib.parse import parse_qs, urlparse
from xml.etree import ElementTree
from aio_pika.abc import AbstractRobustChannel, AbstractRobustConnection, AbstractRobustExchange, AbstractExchange
from aiohttp import ClientSession, ClientResponse
from dataclasses import dataclass
from fastapi import HTTPException
from starlette.responses import FileResponse
from amqp.adapter.data_response_factory import DataResponseFactory
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import DataMessage, DataResponse
from amqp.adapter.multipart_parser import CleverMultiPartParser, process_form_data
from amqp.adapter.file_uploader import publish_file_to_rabbitmq
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
from IPython.core import debugger
debug = debugger.Pdb().set_trace
@dataclass
class AMQMessage(DataMessage):
"""
Expand the core DataMessage type to include the channel, which is required to create a dedicated queue
in case the response should initiate file download.
"""
def __init__(self, data_message: DataMessage, connection: AbstractRobustConnection):
# Copy all fields from DataMessage
self.magic = data_message.magic
self.id = data_message.id
self.method = data_message.method
self.domain = data_message.domain
self.path = data_message.path
self.headers = data_message.headers
self.trace_info = data_message.trace_info
self.base64_body = data_message.base64_body
# Add the new field
self.connection = connection
self.extra_data = data_message.extra_data if hasattr(data_message, 'extra_data') else {}
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
def parse_request_body(message: DataMessage):
"""
Parses the HTTP POST request body based on the MIME type.
This is typically called fom the concrete CleverThis Service to extract parameters from the message body.
Args:
message (DataMessage): Contains raw request and MIME headers.
Returns:
dict: Parsed data as a dictionary.
"""
_content_type = message.headers.get('Content-Type', '')
if not _content_type:
raise ValueError("Content-Type header is required")
content_type = _content_type[0]
# Parse JSON
if content_type == "application/json":
try:
return json.loads(message.body())
except json.JSONDecodeError as e:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {_tb.filename}, line {_tb.lineno}, function '{_tb.name}': {e}")
raise ValueError(f"Invalid JSON: {e}")
# Parse URL-encoded form data
elif content_type == "application/x-www-form-urlencoded":
try:
_form_data: dict = {k: v[0] if len(v) == 1 else v for k, v in parse_qs(message.body()).items()}
if len(_form_data) == 0 and len(message.body()) > 0:
_form_data = json.loads(message.body())
if _form_data['files'] and _form_data['form']:
_form_data = process_form_data(_form_data)
return _form_data
except Exception as e:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {_tb.filename}, line {_tb.lineno}, function '{_tb.name}': {e}")
raise ValueError(f"Invalid URL-encoded form data: {e}")
# Parse multipart/form-data
elif content_type.startswith("multipart/form-data"):
try:
parser: CleverMultiPartParser = CleverMultiPartParser(message)
return parser.form_data
except Exception as e:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {_tb.filename}, line {_tb.lineno}, function '{_tb.name}': {e}")
raise ValueError(f"Invalid multipart/form-data: {e}")
# Parse plain text
elif content_type == "text/plain":
return {"text": message.body()}
# Parse XML
elif content_type == "application/xml":
try:
_root = ElementTree.fromstring(message.body())
return {elem.tag: elem.text for elem in _root}
except ElementTree.ParseError as e:
raise ValueError(f"Invalid XML: {e}")
else:
raise ValueError(f"Unsupported Content-Type: {content_type}")
async def _convert_file_response(
obj: FileResponse,
connection: AbstractRobustConnection,
loop: AbstractEventLoop | None
) -> str:
"""
Converts a FileResponse object to a JSON string containing the filename and stream name.
The file content itself is transmitted through the named stream (which is implemented as temporary dedicated queue.
"""
async def _wrap_amq_api_calls(
connection: AbstractRobustConnection
) -> tuple[AbstractRobustChannel, AbstractExchange, str]:
# Wrap calls to RabbitMQ API to make it Awaitable, so it can be run thread-safe with correct event loop
# This is needed because the RabbitMQ API is not thread-safe and needs to be called in the context of
# the event loop the first RabbitMQ connection was created with.
_channel = await connection.channel()
_exchange_name: str = AMQ_REPLY_TO_EXCHANGE
_exchange = await _channel.get_exchange(name=_exchange_name, ensure=True)
_queue = await _channel.declare_queue(exclusive=False) # Declare a unique, non-exclusive queue
_queue_name = _queue.name
await _queue.bind(
exchange=_exchange_name,
routing_key=_queue_name, # Use queue name as routing key
)
logging.debug(f" [*] Publishing to exchange: {_exchange_name}, q: {_queue_name}")
return _channel, _exchange, _queue_name
# All calls to RabbitMQ API have to be called threadsafe, because those appear to be attached to parent
# event loop, and need to execute at that event loop, not the current one
_future = asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop)
while not _future.done():
await asyncio.sleep(0.1) # allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
_channel, _exchange, _routing_key = _future.result()
# publish_file also needs RabbitMQ API, ensure the loop is passed as argument
(_stream_name, _file_size) = await publish_file_to_rabbitmq(_exchange, obj.path, _routing_key, loop)
# as above, but luckily this time we don't need to wait for the result
_future = asyncio.run_coroutine_threadsafe(_channel.close(), loop)
return json.dumps({'files': [
{'filename': obj.filename,
'mediaType': 'application/octet-stream',
'streamName': _stream_name,
'fileSize': _file_size
}]
})
async def call_cleverthis_api(
message: AMQMessage,
args: Any,
api_method: Callable[[Any, Any], Coroutine],
success_code: int,
loop: AbstractEventLoop | None = None
) -> DataResponse:
"""
Handles POST and PUT requests, which typically contain possibly complex payload data, passed as a dictionary.
Invokes the provided API method with the given arguments and authenticated user,
and returns the response as DataResponse.
:param message: DataMessage containing the details of the call and the payload
: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
:return: DataResponse class
"""
try:
_verified_args = {}
_sig = inspect.signature(api_method)
_required_args = list(_sig.parameters.values())
# _required_args = getattr(api_method, '__annotations__', {})
for arg in _required_args:
if arg.name in args:
# convert to another type if arg is not a str
if arg.annotation == int:
_verified_args[arg.name] = int(args[arg.name])
elif arg.annotation == float:
_verified_args[arg.name] = float(args[arg.name])
# TODO: more?
else:
_verified_args[arg.name] = args[arg.name]
else:
logging.info(f"{api_method.__name__}: Missing required argument: {arg.name}")
logging.info(
f" [*] INVOKE ENDPOINT: {api_method.__name__}, ARGS: {', '.join(f'{k}={v}' for k, v in _verified_args.items())}")
_result: Any = await api_method(**_verified_args)
logging.info(f" [*] ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}")
if isinstance(_result, FileResponse):
_json_result, _content_type = await _convert_file_response(
_result, message.connection, loop
), 'application/octet-stream'
else:
_json_result = _result
_content_type = 'application/json'
_binary_result = _json_result.encode('utf-8') if hasattr(
_json_result, 'encode'
) else _json_result.body if hasattr(
_json_result, 'body'
) else _json_result.read() if hasattr(_json_result, 'read') else _json_result
logging.info(f" [x] ALL DONE in call_cleverthis_api id:{message.id}, resp code:{success_code}")
return DataResponseFactory.of(message.id, success_code, _content_type, _binary_result, message.trace_info)
except HTTPException as http_error:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {_tb.filename}, line {_tb.lineno}, function '{_tb.name}': {http_error}")
return DataResponseFactory.of(message.id, http_error.status_code, "text/plain",
str(http_error.detail).encode('utf-8'), message.trace_info)
except Exception as error:
_tb = traceback.extract_tb(sys.exc_info()[2])[-1]
logging.error(f" [E] Error in {_tb.filename}, line {_tb.lineno}, function '{_tb.name}': {error}")
return DataResponseFactory.of(message.id, 500, "text/plain", str(error).encode('utf-8'), message.trace_info)
# ============= 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.
IMPORTANT: This class is not intended to be used directly. It should be subclassed for each CleverThis service
and override the methods to handle the specific API calls.
"""
def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None):
"""
Initializes the CleverThisServiceAdapter with the given AMQ configuration and optional HTTP session.
The HTTP session is for the case when the adapter uses loose coupling with the CleverThis service via
HTTP REST API.
"""
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
self.loop: AbstractEventLoop | None = None
self.require_authenticated_user = eval(
self.amq_configuration.amq_adapter.require_authenticated_user.capitalize()) if self.amq_configuration.amq_adapter.require_authenticated_user.strip() else True
def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
"""
Pull the value of the Content-Type header from the message headers.
:param message_headers: The headers of the message.
return: The content type of the message.
"""
for header, values in message_headers.items():
if header.lower() == 'content-type':
return values[0].split(';')[0]
return 'application/json'
async def get_current_user(self, token: str) -> object:
"""
To Be Overriden in subclass for given CleverThis Service, to return the current user based on the token,
in format appropriate for the service.
:param token: The token to be used for authentication.
:return: A user object or None if the user is not authenticated.
"""
return None
async def on_message(self, message: AMQMessage) -> DataResponse:
"""
Called when a message is received from the AMQP.
:param message:
:return: AMQP response class with operation status code and optional error details.
"""
_auth_user: Any | None = None
_amq_response: Any | None = None
logging.debug(
f' [*] on_message: _auth_user={_auth_user}, require_authenticated_user={self.require_authenticated_user}')
if self.require_authenticated_user:
_auth: str = message.headers.get('Authorization', '')
logging.info(f"Auth: {_auth}")
if isinstance(_auth, List):
_auth = _auth[0]
if 'Bearer' in _auth:
try:
token = _auth.split(' ')[1]
logging.warn(f'Token: {token}')
_auth_user = await self.get_current_user(token)
except Exception as error:
logging.error(f" [E] on_message: Error getting user: {error}")
if _auth_user or not self.require_authenticated_user or message.path.endswith('/login'):
method = message.method.lower()
if method == 'get':
_amq_response = await self.get(message, _auth_user)
elif method == 'put':
_amq_response = await self.put(message, _auth_user)
elif method == 'post':
_amq_response = await self.post(message, _auth_user)
elif method == 'head':
_amq_response = await self.head(message, _auth_user)
elif method == 'delete':
_amq_response = await self.delete(message, _auth_user)
else:
raise ValueError(f"Unexpected HTTP method: {message.method}")
else:
_amq_response = DataResponseFactory.of(message.id, 401, "text/plain", b"Unauthorized", message.trace_info)
logging.info(f" [x] ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}")
return _amq_response
# ==================================================================================================
# ===================== 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: DataMessage, auth_user: Any) -> DataResponse:
"""
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: DataResponse to be sent back to the API Gateway via AMQP
"""
# 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 DataResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
await _http_resp.read(),
message.trace_info)
return DataResponseFactory.of(message.id, 404, "text/plain",
str("Destination location does not exists").encode('utf-8'), message.trace_info)
async def put(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
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: DataResponse to be sent back to the API Gateway via AMQP
"""
return await self.handle_possible_form(
message,
self.session.put(
f"http://{self.service_host}:{self.service_port}{message.path}",
data=message.body,
headers={**message.headers, **message.trace_info},
),
auth_user
)
async def post(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
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: DataResponse to be sent back to the API Gateway via AMQP
"""
return await self.handle_possible_form(
message,
self.session.post(
f"http://{self.service_host}:{self.service_port}{message.path}",
data=message.body,
headers={**message.headers, **message.trace_info},
),
auth_user
)
async def handle_possible_form(self, message: DataMessage, request_coroutine, auth_user: Any) -> DataResponse:
_args = parse_request_body(message)
return await request_coroutine
async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
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: DataResponse to be sent back to the API Gateway via AMQP
"""
raise NotImplementedError
async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse:
"""
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: DataResponse to be sent back to the API Gateway via AMQP
"""
raise NotImplementedError
Reference in New Issue View Git Blame Copy Permalink