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

Fixes from testing

Browse Source
This commit is contained in:
Stanislav Hejny
2025-04-10 21:38:41 +01:00
parent 63db3c0758
commit 79ed698ac9
4 changed files with 70 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amqp/adapter/cleverthis_service_adapter.py
+8 -6
View File
@@ -153,12 +153,14 @@ async def call_cleverthis_api(
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)}")
_json_result, _content_type = await convert_file_response(_result, message.channel), 'application/octet-stream' \
if isinstance(_result, FileResponse) else _result, '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
_json_result, _content_type = await convert_file_response(
_result, message.channel
), 'application/octet-stream' if isinstance(_result, FileResponse) else _result, '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
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]
amqp/service/amq_message_handler.py
+3 -1
View File
@@ -178,8 +178,10 @@ class DataMessageHandler:
last_data_message_time = time.time()
# Call the async on_message method. This shall also ACK the message after processing.
a_message = AMQMessage(amq_message, self.rabbit_mq_client.channel)
print(f" [*] ######[{message.delivery_tag}] about to call asyncio.run(on_message) with AMQ Message: {a_message}")
response: DataResponse = asyncio.run(
self.service_adapter.on_message(AMQMessage(amq_message, self.rabbit_mq_client.channel))
self.service_adapter.on_message(a_message)
)
try:
logging.info(" [*] ######[%s] AMQ Message Reply.1, To=%s, msgId=%s",
cleverswarm/clever_swarm_adapter.py
+58 -46
View File
@@ -11,6 +11,7 @@ from collections import defaultdict
from typing import Optional, List, Dict
from aiohttp import ClientSession
from fastapi.routing import APIRoute
from fastapi.security import OAuth2PasswordRequestForm
from starlette.datastructures import UploadFile, Headers
# CleverSwarm specific imports
@@ -21,17 +22,15 @@ 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, \
parse_get_url, call_cleverthis_api
from amqp.adapter.multipart_parser import CleverMultiPartParser
parse_get_url, call_cleverthis_api, parse_request_body, AMQMessage
from amqp.config.amq_configuration import AMQConfiguration
from amqp.model.model import AMQMessage, AMQResponse
from amqp.model.model import AMQResponse
from amqp.service.amq_service import AMQService
# Suffix for the method name with serialization wrapper, prefer use this one over the orig endpoint
PREFERRED_SUFFIX = '_json'
def clone_and_adapt_route(original_route: APIRoute) -> APIRoute:
"""
Creates a clone of an APIRoute. It checks if there's preferred endpoint with json serialization.
@@ -80,9 +79,9 @@ def clone_and_adapt_route(original_route: APIRoute) -> APIRoute:
def get_prefered_endpoint(module_name: str, function_name: str) -> Optional[callable]:
"""Get the function if it exists in the module, else return None."""
try:
module = importlib.import_module(module_name)
prefered_endpoint = getattr(module, function_name)
return prefered_endpoint if callable(prefered_endpoint) else None
_module = importlib.import_module(module_name)
_prefered_endpoint = getattr(_module, function_name)
return _prefered_endpoint if callable(_prefered_endpoint) else None
except Exception:
return None
@@ -92,13 +91,13 @@ def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRout
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))
_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)
return dict(_method_to_routes)
class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
@@ -109,8 +108,8 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None):
super().__init__(amq_configuration, session)
self.routes = group_routes_by_method(api_routes=api.api_router.routes)
amq_service: AMQService = AMQService(amq_configuration, self)
amq_service.rabbit_mq_client.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:
"""
@@ -182,19 +181,19 @@ 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
"""
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
_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_api(
message,
args,
api_method=route.endpoint,
success_code=route.status_code
_args,
api_method=_route.endpoint,
success_code=_route.status_code
)
return AMQResponseFactory.of(message.id, 404, "text/plain",
@@ -209,34 +208,47 @@ class CleverSwarmAmqpAdapter(CleverThisServiceAdapter):
: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:
_form_data = parse_request_body(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 \
_form_data.update(_match.groupdict())
_form_data.update(_query_args)
for _body_param in _route.dependant.body_params:
if _body_param.name in _form_data:
_file_data = _form_data[_body_param.name]
actual_filepath = message.extra_data[_body_param.name]
_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]
size=_file_data['size'],
filename=_file_data['filename'],
headers=Headers({"Content-Type": _file_data['mediaType']})
) or _form_data[_body_param.name]
_verified_args = {}
_required_args = getattr(_route.endpoint, '__annotations__', {})
for arg in _required_args:
if arg in _form_data:
_verified_args[arg] = _form_data[arg]
else:
# needed for /login input
if arg == 'authenticated_user':
_form_data['authenticated_user'] = auth_user
if _required_args[arg].__name__ == 'OAuth2PasswordRequestForm':
try:
_verified_args[arg] = OAuth2PasswordRequestForm(**_form_data)
except Exception as e:
print(f"Error: {e}")
return await call_cleverthis_api(
message,
parser.form_data,
api_method=route.endpoint,
success_code=route.status_code
_verified_args,
api_method=_route.endpoint,
success_code=_route.status_code if _route.status_code else 200
)
return AMQResponseFactory.of(message.id, 404, "text/plain",
pyproject.toml
+1 -1
View File
@@ -8,7 +8,7 @@ packages = ["amqp","amqp.adapter","amqp.config","amqp.model","amqp.rabbitmq","am
[project]
name = "amq_adapter"
version = "0.2.16"
version = "0.2.18"
description = "The Python implementation of the AMQ Adaptor for CleverMicro"
readme = "README.md"
license = { text = "Apache License 2.0" }