Stanislav Hejny
425 lines
18 KiB
Python
425 lines
18 KiB
Python
"""
|
|
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
|
|
from asyncio import AbstractEventLoop
|
|
from typing import Dict, List, Callable, Coroutine, Any
|
|
|
|
from aio_pika.abc import (
|
|
AbstractRobustChannel,
|
|
AbstractRobustConnection,
|
|
AbstractExchange,
|
|
)
|
|
from aiohttp import ClientSession, ClientResponse
|
|
from dataclasses import dataclass
|
|
from fastapi import HTTPException
|
|
from starlette.responses import FileResponse
|
|
|
|
from amqp.adapter.data_parser import AMQDataParser
|
|
from amqp.adapter.data_response_factory import DataResponseFactory
|
|
from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
|
|
from amqp.adapter.logging_utils import logging_error, logging_debug, logging_info
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
from amqp.model.model import DataMessage, DataResponse
|
|
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
|
|
from amqp.router.utils import await_result
|
|
|
|
|
|
@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 {}
|
|
)
|
|
|
|
|
|
async def _convert_file_response(
|
|
obj: FileResponse,
|
|
connection: AbstractRobustConnection,
|
|
file_handler: FileHandler,
|
|
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
|
|
# allow coroutine to execute. calling _future.result() will block the entire thread and the coroutine will never execute
|
|
_channel, _exchange, _routing_key = await await_result(
|
|
asyncio.run_coroutine_threadsafe(_wrap_amq_api_calls(connection), loop)
|
|
)
|
|
# publish_file also needs RabbitMQ API, ensure the loop is passed as argument
|
|
(_stream_name, _file_size) = await file_handler.publish_file(
|
|
_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,
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
|
|
def is_likely_json(text: str) -> bool:
|
|
"""Quick test to see if should convert the response string to JSON format."""
|
|
_text = text.strip()
|
|
return (_text.startswith("{") and _text.endswith("}")) or (
|
|
_text.startswith("[") and _text.endswith("]")
|
|
)
|
|
|
|
|
|
# ============= 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 = (
|
|
self.amq_configuration.amq_adapter.require_authenticated_user
|
|
)
|
|
|
|
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: require_authenticated_user={self.require_authenticated_user}"
|
|
)
|
|
if self.require_authenticated_user:
|
|
_auth: str = message.headers.get("Authorization", "")
|
|
logging_debug(f"Auth: {_auth}")
|
|
if isinstance(_auth, List):
|
|
_auth = _auth[0]
|
|
if "Bearer" in _auth:
|
|
try:
|
|
token = _auth.split(" ")[1]
|
|
logging_info(f"Token: {token}")
|
|
_auth_user = await self.get_current_user(token)
|
|
except Exception as error:
|
|
logging_error(f"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"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 = AMQDataParser.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
|
|
|
|
@staticmethod
|
|
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])
|
|
elif arg.annotation == bool:
|
|
_verified_args[arg.name] = bool(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,
|
|
file_handler=StreamingFileHandler(),
|
|
loop=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"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:
|
|
_content = str(http_error.detail)
|
|
if not is_likely_json(_content):
|
|
_content = '{"error": "' + _content + '"}'
|
|
logging_error(f"Service REST endpoint invocation failed: {http_error}")
|
|
return DataResponseFactory.of(
|
|
message.id,
|
|
http_error.status_code,
|
|
"application/json",
|
|
_content.encode("utf-8"),
|
|
message.trace_info,
|
|
)
|
|
except Exception as error:
|
|
logging_error(f"Service endpoint invocation failed: {error}")
|
|
_content = str(error)
|
|
if not is_likely_json(_content):
|
|
_content = '{"error": "' + _content + '"}'
|
|
return DataResponseFactory.of(
|
|
message.id,
|
|
500,
|
|
"application/json",
|
|
_content.encode("utf-8"),
|
|
message.trace_info,
|
|
)
|