#23 refactor to use shared code between services
This commit is contained in:
Stanislav Hejny
@@ -163,10 +163,7 @@ class BackpressureHandler:
|
||||
CPU_IDLE, CPU_ACTIVE, CPU_OVERLOAD = 0, 1, 2
|
||||
IDLE_THRESHOLD, OVERLOAD_THRESHOLD = 15, 90
|
||||
self.average_cpu_usage = RunningAverage(
|
||||
int(
|
||||
max(_overload_duration_millis, _idle_duration_millis)
|
||||
// _monitor_interval
|
||||
)
|
||||
int(max(_overload_duration_millis, _idle_duration_millis) // _monitor_interval)
|
||||
)
|
||||
|
||||
async def _backpressure_monitor():
|
||||
@@ -283,15 +280,11 @@ class BackpressureHandler:
|
||||
if not self.exchange:
|
||||
|
||||
async def _wrap_rabbit_mq_api_init(channel):
|
||||
_exchange = await channel.get_exchange(
|
||||
name="cleverthis.clevermicro.management"
|
||||
)
|
||||
_exchange = await channel.get_exchange(name="cleverthis.clevermicro.management")
|
||||
return _exchange
|
||||
|
||||
self.exchange = await await_result(
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
_wrap_rabbit_mq_api_init(self.channel), self.loop
|
||||
)
|
||||
asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api_init(self.channel), self.loop)
|
||||
)
|
||||
|
||||
if self.exchange:
|
||||
@@ -319,6 +312,4 @@ class BackpressureHandler:
|
||||
return True
|
||||
return False
|
||||
|
||||
await await_future(
|
||||
asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop)
|
||||
)
|
||||
await await_future(asyncio.run_coroutine_threadsafe(_wrap_rabbit_mq_api(), self.loop))
|
||||
|
||||
@@ -6,8 +6,10 @@ Methods here will invoke the actual CleverThis Service code, which are passed as
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
import pathlib
|
||||
import re
|
||||
from asyncio import AbstractEventLoop
|
||||
from dataclasses import dataclass
|
||||
from collections import defaultdict
|
||||
from typing import Any, Callable, Coroutine, Dict, List
|
||||
|
||||
from aio_pika.abc import (
|
||||
@@ -16,44 +18,24 @@ from aio_pika.abc import (
|
||||
AbstractRobustConnection,
|
||||
)
|
||||
from aiohttp import ClientResponse, ClientSession
|
||||
from fastapi import HTTPException
|
||||
from fastapi import HTTPException, UploadFile
|
||||
from fastapi.routing import APIRoute
|
||||
from fastapi.security import OAuth2PasswordRequestForm
|
||||
from pydantic import BaseModel
|
||||
from starlette.datastructures import Headers
|
||||
from starlette.responses import FileResponse
|
||||
|
||||
from amqp.adapter import serializer
|
||||
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_debug, logging_error, logging_info
|
||||
from amqp.config.amq_configuration import AMQConfiguration
|
||||
from amqp.model.model import AMQErrorMessage, DataMessage, DataResponse
|
||||
from amqp.model.model import AMQErrorMessage, AMQMessage, 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,
|
||||
@@ -62,15 +44,16 @@ async def _convert_file_response(
|
||||
) -> 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.
|
||||
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.
|
||||
# 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 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)
|
||||
@@ -85,9 +68,10 @@ async def _convert_file_response(
|
||||
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
|
||||
# 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)
|
||||
)
|
||||
@@ -122,19 +106,22 @@ def is_likely_json(text: str) -> bool:
|
||||
# ============= 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.
|
||||
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
|
||||
self,
|
||||
amq_configuration: AMQConfiguration,
|
||||
routes: List[APIRoute],
|
||||
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.
|
||||
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
|
||||
@@ -144,6 +131,79 @@ class CleverThisServiceAdapter:
|
||||
self.require_authenticated_user = (
|
||||
self.amq_configuration.amq_adapter.require_authenticated_user
|
||||
)
|
||||
self.routes = self.group_routes_by_method(api_routes=routes)
|
||||
|
||||
def get_endpoint(self, endpoint: Any) -> Any:
|
||||
"""
|
||||
Optionally, Service can override this method to alter the endpoint mapping.
|
||||
This is useful for example when the CleverThis service has a JSON wrapper
|
||||
around the endpoint, like CleverSwarm, which might be the preferred endpoint to call.
|
||||
|
||||
:param endpoint: The original endpoint to be called.
|
||||
:return: The actual endpoint to be called.
|
||||
"""
|
||||
return endpoint
|
||||
|
||||
def clone_and_adapt_route(self, original_route: APIRoute) -> APIRoute:
|
||||
"""
|
||||
Creates a clone of APIRoute. Checks if there's preferred endpoint with json serialization.
|
||||
It also adjusts the path regex to ensure it matches the path in different API versions.
|
||||
"""
|
||||
|
||||
# Need to adapt also the path regex to ensure it matches the path in different API versions
|
||||
# Ensures the leading '^' is followed by any match and there's no '/' before final '$'
|
||||
_new_regex_string = "^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$"
|
||||
_new_route: APIRoute = APIRoute(
|
||||
path=original_route.path,
|
||||
endpoint=self.get_endpoint(
|
||||
original_route.endpoint
|
||||
), # Service can alter the endpoint mapping here
|
||||
response_model=original_route.response_model,
|
||||
status_code=original_route.status_code,
|
||||
tags=original_route.tags.copy() if original_route.tags else None,
|
||||
dependencies=(
|
||||
original_route.dependencies.copy() if original_route.dependencies else None
|
||||
),
|
||||
summary=original_route.summary,
|
||||
description=original_route.description,
|
||||
response_description=original_route.response_description,
|
||||
responses=(original_route.responses.copy() if original_route.responses else None),
|
||||
deprecated=original_route.deprecated,
|
||||
name=original_route.name,
|
||||
methods=original_route.methods.copy() if original_route.methods else None,
|
||||
operation_id=original_route.operation_id,
|
||||
response_model_include=original_route.response_model_include,
|
||||
response_model_exclude=original_route.response_model_exclude,
|
||||
response_model_by_alias=original_route.response_model_by_alias,
|
||||
response_model_exclude_unset=original_route.response_model_exclude_unset,
|
||||
response_model_exclude_defaults=original_route.response_model_exclude_defaults,
|
||||
response_model_exclude_none=original_route.response_model_exclude_none,
|
||||
include_in_schema=original_route.include_in_schema,
|
||||
response_class=original_route.response_class,
|
||||
dependency_overrides_provider=original_route.dependency_overrides_provider,
|
||||
callbacks=(original_route.callbacks.copy() if original_route.callbacks else None),
|
||||
openapi_extra=(
|
||||
original_route.openapi_extra.copy() if original_route.openapi_extra else None
|
||||
),
|
||||
generate_unique_id_function=original_route.generate_unique_id_function,
|
||||
)
|
||||
_new_route.path_regex = re.compile(_new_regex_string)
|
||||
return _new_route
|
||||
|
||||
def group_routes_by_method(self, api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]:
|
||||
"""
|
||||
Groups API routes by their HTTP method. This is important to ensure that 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 (
|
||||
isinstance(_route, APIRoute) and _route.methods
|
||||
): # Ensure methods exist (should always be true for valid routes)
|
||||
for method in _route.methods:
|
||||
_method_to_routes[method].append(self.clone_and_adapt_route(_route))
|
||||
|
||||
return dict(_method_to_routes)
|
||||
|
||||
def get_content_type(self, message_headers: Dict[str, List[str]]) -> str:
|
||||
"""
|
||||
@@ -158,9 +218,11 @@ class CleverThisServiceAdapter:
|
||||
|
||||
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.
|
||||
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
|
||||
@@ -174,11 +236,9 @@ class CleverThisServiceAdapter:
|
||||
_auth_user: Any | None = None
|
||||
_amq_response: Any | None = None
|
||||
|
||||
logging_debug(
|
||||
f"on_message: require_authenticated_user={self.require_authenticated_user}"
|
||||
)
|
||||
logging_debug(f"on_message: require_authenticated_user={self.require_authenticated_user}")
|
||||
if self.require_authenticated_user:
|
||||
_auth: str = message.headers.get("Authorization", "")
|
||||
_auth: str = message.headers().get("Authorization", "")
|
||||
logging_debug(f"Auth: {_auth}")
|
||||
if isinstance(_auth, List):
|
||||
_auth = _auth[0]
|
||||
@@ -190,12 +250,8 @@ class CleverThisServiceAdapter:
|
||||
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 _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":
|
||||
@@ -207,117 +263,296 @@ class CleverThisServiceAdapter:
|
||||
elif method == "delete":
|
||||
_amq_response = await self.delete(message, _auth_user)
|
||||
else:
|
||||
raise ValueError(f"Unexpected HTTP method: {message.method}")
|
||||
raise ValueError(f"Unexpected HTTP method: {message.method()}")
|
||||
else:
|
||||
_amq_response = DataResponseFactory.of_error_message(
|
||||
message.id,
|
||||
message.id(),
|
||||
AMQErrorMessage(401, "Unauthorized", "Unauthorized"),
|
||||
message.trace_info,
|
||||
message.trace_info(),
|
||||
)
|
||||
logging_info(
|
||||
f"ALL DONE in on_message id:{message.id}, resp code:{_amq_response.response_code}"
|
||||
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_error_message(
|
||||
message.id,
|
||||
AMQErrorMessage(404, "Not Found", "Destination location does not exists"),
|
||||
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._authorize_forward_request(
|
||||
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._authorize_forward_request(
|
||||
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 _authorize_forward_request(
|
||||
self, message: DataMessage, request_coroutine, auth_user: Any
|
||||
) -> DataResponse:
|
||||
"""
|
||||
This method is used to authorize the request and forward it to the CleverThis Service API.
|
||||
However, it is only used for loose / 3rd party mode of coupling with the Service
|
||||
Used only for loose / 3rd party mode of coupling with the Service
|
||||
"""
|
||||
# AMQDataParser.parse_request_body(message)
|
||||
# TODO: implement authorization logic here - ensure the security related headers are set
|
||||
# This depends on what kind of security is required by the CleverThis service.
|
||||
# However at this time there is no loosely coupled Service.
|
||||
|
||||
return await request_coroutine
|
||||
|
||||
async def head(self, message: DataMessage, auth_user: Any) -> DataResponse:
|
||||
# ==============================================================================================
|
||||
# ===================== 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: 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
|
||||
Handles the GET requests for the CleverThis API. In tight coupling (self.session is None),
|
||||
use path, path params and query string to determine the endpoint method to invoke.
|
||||
In loose coupling (self.session is not None), forward the request as REST request using
|
||||
the session, i.e. 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
|
||||
"""
|
||||
|
||||
if self.session is None:
|
||||
return await self.handle_no_body_payload(message, auth_user, "GET")
|
||||
|
||||
# Forwards as REST - 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}"
|
||||
)
|
||||
_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(),
|
||||
)
|
||||
|
||||
async def put(self, message: AMQMessage, auth_user: Any) -> DataResponse:
|
||||
"""
|
||||
Handles the PUT requests for the CleverThis API. Uses path, path params, query string and
|
||||
body payload to determine the endpoint method to invoke and its input arguments
|
||||
In loose coupling (self.session is not None), 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
|
||||
"""
|
||||
if self.session is None:
|
||||
return await self.handle_body_payload(message, auth_user, "PUT")
|
||||
|
||||
# Forwards as REST - try to invoke the service directly on this path
|
||||
return await self._authorize_forward_request(
|
||||
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: AMQMessage, auth_user: Any) -> DataResponse:
|
||||
"""
|
||||
Handles the POST requests for the CleverThis API. Uses path, path params, query string and
|
||||
body payload to determine the endpoint method to invoke and its input arguments
|
||||
In loose coupling (self.session is not None), attempt to call the REST endpoint directly
|
||||
using HTTP protocol.
|
||||
|
||||
:param message: message from 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
|
||||
"""
|
||||
if self.session is None:
|
||||
return await self.handle_body_payload(message, auth_user, "POST")
|
||||
|
||||
return await self._authorize_forward_request(
|
||||
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 head(self, message: AMQMessage, auth_user: Any) -> DataResponse:
|
||||
"""
|
||||
Handles the HEAD requests for the CleverThis API. In tight coupling (self.session is None),
|
||||
use path, path params and query string to determine the endpoint method to invoke.
|
||||
In loose coupling (self.session is not None), HEAD call is not supported.
|
||||
|
||||
:param message: message from AMQP containing the HEAD request, including body and headers
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchema
|
||||
|
||||
:return: DataResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
if self.session is None:
|
||||
return await self.handle_no_body_payload(message, auth_user, "HEAD")
|
||||
|
||||
raise NotImplementedError
|
||||
|
||||
async def delete(self, message: DataMessage, auth_user: Any) -> DataResponse:
|
||||
async def delete(self, message: AMQMessage, 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
|
||||
Handles the DELETE requests for the CleverThis API.
|
||||
|
||||
:param message: message from AMQP that contains the DELETE request, including body & headers
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchema
|
||||
|
||||
:return: DataResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
if self.session is None:
|
||||
return await self.handle_no_body_payload(message, auth_user, "DELETE")
|
||||
|
||||
return await self._authorize_forward_request(
|
||||
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 options(self, message: AMQMessage, auth_user: Any) -> DataResponse:
|
||||
"""
|
||||
Handles the OPTIONS requests. In tight coupling (self.session is None),
|
||||
use path, path params and query string to determine the endpoint method to invoke.
|
||||
In loose coupling (self.session is not None), forward the request as REST request using
|
||||
the session, i.e. attempt to call the REST endpoint directly using HTTP protocol.
|
||||
|
||||
:param message: message from AMQP that contains the OPTIONS request, including headers
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchema
|
||||
|
||||
:return: DataResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
if self.session is None:
|
||||
return await self.handle_no_body_payload(message, auth_user, "OPTIONS")
|
||||
|
||||
raise NotImplementedError
|
||||
|
||||
async def patch(self, message: AMQMessage, auth_user: Any) -> DataResponse:
|
||||
"""
|
||||
Handles the PATCH requests for the CleverBRAG API.
|
||||
|
||||
:param message: message from AMQP that contains the PATCH request, including body & headers
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchema
|
||||
|
||||
:return: DataResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
if self.session is None:
|
||||
return await self.handle_no_body_payload(message, auth_user, "PATCH")
|
||||
|
||||
return await self._authorize_forward_request(
|
||||
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,
|
||||
)
|
||||
|
||||
# ==============================================================================================
|
||||
# ================= C l e v e r S w a r m p a y l o a d h a n d l e r s ==================
|
||||
# ==============================================================================================
|
||||
|
||||
async def handle_no_body_payload(
|
||||
self, message: AMQMessage, auth_user: Any, method: str
|
||||
) -> DataResponse:
|
||||
"""
|
||||
Handles the requests that do not have a body payload.
|
||||
:param message: message from the AMQP that contains the request
|
||||
: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
|
||||
_routes = self.routes.get(method, [])
|
||||
_path, _args = AMQDataParser.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 CleverThisServiceAdapter.call_cleverthis_api(
|
||||
message,
|
||||
_args,
|
||||
api_method=_route.endpoint,
|
||||
success_code=_route.status_code,
|
||||
loop=self.loop,
|
||||
)
|
||||
|
||||
return DataResponseFactory.of(
|
||||
message.id(),
|
||||
404,
|
||||
"text/plain",
|
||||
str(f"Destination location [{message.path()}] does not exists").encode("utf-8"),
|
||||
message.trace_info(),
|
||||
)
|
||||
|
||||
async def handle_body_payload(
|
||||
self, message: AMQMessage, auth_user: Any, method: str
|
||||
) -> DataResponse:
|
||||
"""
|
||||
Handles the requests that have a body payload, including Multipart data.
|
||||
:param message: message from the AMQP that contains the request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:param method: HTTP method (POST, PUT, etc.)
|
||||
:return: DataResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
_form_data = AMQDataParser.parse_request_body(message)
|
||||
_routes = self.routes.get(method, [])
|
||||
_path, _query_args = AMQDataParser.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
|
||||
_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=_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":
|
||||
_verified_args["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 CleverThisServiceAdapter.call_cleverthis_api(
|
||||
message,
|
||||
_verified_args,
|
||||
api_method=_route.endpoint,
|
||||
success_code=_route.status_code if _route.status_code else 200,
|
||||
loop=self.loop,
|
||||
)
|
||||
|
||||
return DataResponseFactory.of(
|
||||
message.id(),
|
||||
404,
|
||||
"text/plain",
|
||||
str(f"Destination location [{message.path()}] does not exists").encode("utf-8"),
|
||||
message.trace_info(),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def call_cleverthis_api(
|
||||
@@ -328,13 +563,15 @@ class CleverThisServiceAdapter:
|
||||
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.
|
||||
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:
|
||||
@@ -350,23 +587,19 @@ class CleverThisServiceAdapter:
|
||||
elif arg.annotation == float:
|
||||
_verified_args[arg.name] = float(args[arg.name])
|
||||
elif arg.annotation == bool:
|
||||
_verified_args[arg.name] = serializer.str_to_bool(
|
||||
args[arg.name]
|
||||
)
|
||||
_verified_args[arg.name] = serializer.str_to_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"{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())}"
|
||||
"INVOKE ENDPOINT: {%s}, ARGS: {%s}",
|
||||
api_method.__name__,
|
||||
", ".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)}"
|
||||
)
|
||||
logging_info(f"ENDPOINT RESULT: {api_method.__name__}, RESULT: {str(_result)}")
|
||||
|
||||
if isinstance(_result, FileResponse):
|
||||
_json_result, _content_type = (
|
||||
@@ -391,60 +624,56 @@ class CleverThisServiceAdapter:
|
||||
else (
|
||||
_json_result.body
|
||||
if hasattr(_json_result, "body")
|
||||
else (
|
||||
_json_result.read()
|
||||
if hasattr(_json_result, "read")
|
||||
else _json_result
|
||||
)
|
||||
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}"
|
||||
f"ALL DONE in call_cleverthis_api id:{message.id()}, resp code:{success_code}"
|
||||
)
|
||||
return DataResponseFactory.of(
|
||||
message.id,
|
||||
message.id(),
|
||||
success_code,
|
||||
_content_type,
|
||||
_binary_result,
|
||||
message.trace_info,
|
||||
message.trace_info(),
|
||||
)
|
||||
except HTTPException as http_error:
|
||||
logging_error(f"Service REST endpoint invocation failed: {http_error}")
|
||||
_content = str(http_error.detail)
|
||||
if not is_likely_json(_content):
|
||||
return DataResponseFactory.of_error_message(
|
||||
message.id,
|
||||
message.id(),
|
||||
AMQErrorMessage(
|
||||
http_error.status_code,
|
||||
"Service Invocation Failed",
|
||||
_content,
|
||||
),
|
||||
message.trace_info,
|
||||
message.trace_info(),
|
||||
)
|
||||
return DataResponseFactory.of(
|
||||
message.id,
|
||||
message.id(),
|
||||
http_error.status_code,
|
||||
"application/json",
|
||||
_content.encode("utf-8"),
|
||||
message.trace_info,
|
||||
message.trace_info(),
|
||||
)
|
||||
except Exception as error:
|
||||
logging_error(f"Service endpoint invocation failed: {error}")
|
||||
_content = str(error)
|
||||
if not is_likely_json(_content):
|
||||
return DataResponseFactory.of_error_message(
|
||||
message.id,
|
||||
message.id(),
|
||||
AMQErrorMessage(
|
||||
500,
|
||||
"Service Invocation Failed",
|
||||
_content,
|
||||
),
|
||||
message.trace_info,
|
||||
message.trace_info(),
|
||||
)
|
||||
return DataResponseFactory.of(
|
||||
message.id,
|
||||
message.id(),
|
||||
500,
|
||||
"application/json",
|
||||
_content.encode("utf-8"),
|
||||
message.trace_info,
|
||||
message.trace_info(),
|
||||
)
|
||||
|
||||
@@ -5,21 +5,21 @@ methods to create, serialize and deserialize DataMessage objects.
|
||||
import base64
|
||||
import json
|
||||
import re
|
||||
from io import BytesIO
|
||||
from datetime import datetime, timezone
|
||||
from io import BytesIO
|
||||
from typing import Dict, List
|
||||
|
||||
import amqp.adapter.file_handler
|
||||
from amqp.model.model import DataMessage
|
||||
from amqp.model.snowflake_id import SnowflakeId
|
||||
from amqp.adapter.file_handler import FileHandler, StreamingFileHandler
|
||||
from amqp.adapter.serializer import (
|
||||
map_with_list_as_string,
|
||||
map_as_string,
|
||||
map_with_list_as_string,
|
||||
parse_map_list_string,
|
||||
parse_map_string,
|
||||
)
|
||||
from amqp.model.model import DataMessage, DataMessageMagicByte
|
||||
from amqp.model.snowflake_id import SnowflakeId
|
||||
from amqp.router.utils import sanitize_as_rabbitmq_name
|
||||
from amqp.adapter.file_handler import StreamingFileHandler, FileHandler
|
||||
|
||||
|
||||
class DataMessageFactory:
|
||||
@@ -30,7 +30,6 @@ class DataMessageFactory:
|
||||
|
||||
MACHINE_ID_MASK = (1 << MACHINE_ID_BITS) - 1
|
||||
SEQUENCE_MASK = (1 << SEQUENCE_BITS) - 1
|
||||
MAGIC_BYTE_HTTP_REQUEST = "A"
|
||||
|
||||
_instance = None
|
||||
snowflake_epoch = datetime(2024, 10, 1, 0, 0, tzinfo=timezone.utc).timestamp()
|
||||
@@ -100,12 +99,10 @@ class DataMessageFactory:
|
||||
:return: DataMessage object
|
||||
"""
|
||||
serializable_headers = headers.copy() if headers else {}
|
||||
payload = base64.b64encode(
|
||||
message.encode("utf-8") if isinstance(message, str) else message
|
||||
)
|
||||
payload = base64.b64encode(message.encode("utf-8") if isinstance(message, str) else message)
|
||||
trace_info = {}
|
||||
return DataMessage(
|
||||
self.MAGIC_BYTE_HTTP_REQUEST,
|
||||
DataMessageMagicByte.HTTP_REQUEST.value,
|
||||
self.generate_snowflake_id(),
|
||||
method,
|
||||
domain,
|
||||
@@ -122,7 +119,7 @@ class DataMessageFactory:
|
||||
:param message: DataMessage input
|
||||
:return: routing key (as string) compliant to RabbitMQ allowed character set.
|
||||
"""
|
||||
return sanitize_as_rabbitmq_name(f"{message.domain}.{message.path}")
|
||||
return sanitize_as_rabbitmq_name(f"{message.domain()}.{message.path()}")
|
||||
|
||||
@staticmethod
|
||||
def get_timestamp_from_id(_id: SnowflakeId) -> int:
|
||||
@@ -132,9 +129,7 @@ class DataMessageFactory:
|
||||
:return: the message timestamp
|
||||
"""
|
||||
bits = _id.get_bits()
|
||||
low = bits[0] >> (
|
||||
DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS
|
||||
)
|
||||
low = bits[0] >> (DataMessageFactory.SEQUENCE_BITS + DataMessageFactory.MACHINE_ID_BITS)
|
||||
high = bits[1] << (
|
||||
64 - DataMessageFactory.SEQUENCE_BITS - DataMessageFactory.MACHINE_ID_BITS
|
||||
)
|
||||
@@ -149,13 +144,13 @@ class DataMessageFactory:
|
||||
"""
|
||||
return (
|
||||
f"DataMessage{{"
|
||||
f"id={message.id}, "
|
||||
f"method='{message.method}', "
|
||||
f"domain='{message.domain}', "
|
||||
f"path='{message.path}', "
|
||||
f"headers={{{map_with_list_as_string(message.headers)}}}, "
|
||||
f"traceInfo={{{map_as_string(message.trace_info)}}}, "
|
||||
f"base64body={message.base64_body.decode()}"
|
||||
f"id={message.id()}, "
|
||||
f"method='{message.method()}', "
|
||||
f"domain='{message.domain()}', "
|
||||
f"path='{message.path()}', "
|
||||
f"headers={{{map_with_list_as_string(message.headers())}}}, "
|
||||
f"traceInfo={{{map_as_string(message.trace_info())}}}, "
|
||||
f"base64body={message.base64body().decode()}"
|
||||
f"}}"
|
||||
)
|
||||
|
||||
@@ -166,19 +161,19 @@ class DataMessageFactory:
|
||||
:param message: message to serialize
|
||||
:return: messages as bytes
|
||||
"""
|
||||
id_bytes = str(message.id).encode()
|
||||
id_bytes = str(message.id()).encode()
|
||||
prolog = (
|
||||
f"|m={message.method}|d={message.domain}|p={message.path}|h={{{map_with_list_as_string(message.headers)}}}|t={{{map_as_string(message.trace_info)}}}|b="
|
||||
f"|m={message.method()}|d={message.domain()}|p={message.path()}|h={{{map_with_list_as_string(message.headers())}}}|t={{{map_as_string(message.trace_info())}}}|b="
|
||||
).encode("utf-8")
|
||||
header_length = 1 + len(id_bytes) + len(prolog)
|
||||
prolog_len = header_length.to_bytes(2, byteorder="big")
|
||||
msg_length = 2 + header_length + len(message.base64_body)
|
||||
msg_length = 2 + header_length + len(message.base64body())
|
||||
with BytesIO(bytearray(msg_length)) as baos:
|
||||
baos.write(prolog_len)
|
||||
baos.write(DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8"))
|
||||
baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
|
||||
baos.write(id_bytes)
|
||||
baos.write(prolog)
|
||||
baos.write(message.base64_body)
|
||||
baos.write(message.base64body())
|
||||
return baos.getvalue()
|
||||
|
||||
@staticmethod
|
||||
@@ -192,7 +187,7 @@ class DataMessageFactory:
|
||||
metadata = input_bytes[2 : prolog_len + 2].decode()
|
||||
|
||||
pattern = re.compile(
|
||||
f"{DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST}"
|
||||
f"{DataMessageMagicByte.HTTP_REQUEST.value}"
|
||||
r"([0-9A-Fa-f]+)\|"
|
||||
r"m=([^|]*)\|"
|
||||
r"d=([^|]*)\|"
|
||||
@@ -220,7 +215,7 @@ class DataMessageFactory:
|
||||
trace_info = parse_map_string(trace_info_str)
|
||||
|
||||
return DataMessage(
|
||||
DataMessageFactory.MAGIC_BYTE_HTTP_REQUEST,
|
||||
DataMessageMagicByte.HTTP_REQUEST.value,
|
||||
SnowflakeId.from_hex(id_str),
|
||||
method,
|
||||
domain,
|
||||
|
||||
+16
-28
@@ -1,16 +1,15 @@
|
||||
import io
|
||||
import json
|
||||
import python_multipart
|
||||
from tempfile import SpooledTemporaryFile
|
||||
|
||||
from fastapi import UploadFile
|
||||
from python_multipart.multipart import Field, File
|
||||
from typing import Tuple, Dict
|
||||
from typing import Dict, Tuple
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from amqp.adapter.logging_utils import logging_error, logging_debug
|
||||
from amqp.model.model import DataMessage
|
||||
import python_multipart
|
||||
from fastapi import UploadFile
|
||||
from python_multipart.multipart import Field, File
|
||||
|
||||
from amqp.adapter.logging_utils import logging_debug, logging_error
|
||||
from amqp.model.model import AMQMessage
|
||||
|
||||
|
||||
class MultipartFormDataParser:
|
||||
@@ -23,15 +22,14 @@ class MultipartFormDataParser:
|
||||
included, the message is transmitted in the unchanged multipart/form-data format.
|
||||
"""
|
||||
|
||||
def __init__(self, message: DataMessage):
|
||||
def __init__(self, message: AMQMessage):
|
||||
self.form_data: dict = {}
|
||||
self.is_multipart = False
|
||||
self.is_json = False
|
||||
self.is_valid = True
|
||||
# Convert each list of strings to a single string joined by newline, then to bytes
|
||||
_headers = {
|
||||
key: "\n".join(value).encode("utf-8")
|
||||
for key, value in message.headers.items()
|
||||
key: "\n".join(value).encode("utf-8") for key, value in message.headers().items()
|
||||
}
|
||||
try:
|
||||
if (
|
||||
@@ -53,9 +51,7 @@ class MultipartFormDataParser:
|
||||
)
|
||||
self.is_multipart = True
|
||||
except Exception as e:
|
||||
logging_error(
|
||||
f"Error parsing multipart/form-data: {e}. Trying parsing as JSON."
|
||||
)
|
||||
logging_error(f"Error parsing multipart/form-data: {e}. Trying parsing as JSON.")
|
||||
try:
|
||||
self.form_data = json.loads(message.body())
|
||||
self.is_json = True
|
||||
@@ -72,11 +68,8 @@ class MultipartFormDataParser:
|
||||
def on_file(self, file: File):
|
||||
logging_debug(str(file))
|
||||
file.file_object.seek(0)
|
||||
_spooled_file: SpooledTemporaryFile = SpooledTemporaryFile()
|
||||
_spooled_file.write(file.file_object.read())
|
||||
_spooled_file.seek(0)
|
||||
self.form_data[file.field_name.decode("utf-8")] = UploadFile(
|
||||
_spooled_file, size=file.size, filename=file.file_name.decode("utf-8")
|
||||
file.file_object, size=file.size, filename=file.file_name.decode("utf-8")
|
||||
)
|
||||
|
||||
|
||||
@@ -135,13 +128,11 @@ class AMQDataParser:
|
||||
_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()
|
||||
}
|
||||
_simplified_params = {k: v[0] if len(v) == 1 else v for k, v in _query_params.items()}
|
||||
return _parsed.path, _simplified_params
|
||||
|
||||
@staticmethod
|
||||
def parse_request_body(message: DataMessage):
|
||||
def parse_request_body(message: AMQMessage):
|
||||
"""
|
||||
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.
|
||||
@@ -151,7 +142,7 @@ class AMQDataParser:
|
||||
Returns:
|
||||
dict: Parsed data as a dictionary.
|
||||
"""
|
||||
_content_type = message.headers.get("Content-Type", "")
|
||||
_content_type = message.headers().get("Content-Type", "")
|
||||
if not _content_type:
|
||||
raise ValueError("Content-Type header is required")
|
||||
content_type = _content_type[0]
|
||||
@@ -167,15 +158,12 @@ class AMQDataParser:
|
||||
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()
|
||||
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 = MultipartFormDataParser.process_form_data(
|
||||
_form_data
|
||||
)
|
||||
_form_data = AMQDataParser.process_form_data(_form_data)
|
||||
return _form_data
|
||||
except Exception as e:
|
||||
logging_error(f"Invalid URL-encoded form data: {e}")
|
||||
|
||||
@@ -5,13 +5,13 @@ method to create, serialize and deserialize DataResponse message
|
||||
import base64
|
||||
import re
|
||||
from io import BytesIO
|
||||
from amqp.model.model import DataResponse, AMQErrorMessage
|
||||
|
||||
from amqp.adapter.serializer import map_as_string, parse_map_string
|
||||
from amqp.model.model import AMQErrorMessage, DataMessageMagicByte, DataResponse
|
||||
from amqp.model.snowflake_id import SnowflakeId
|
||||
from amqp.adapter.serializer import parse_map_string, map_as_string
|
||||
|
||||
|
||||
class DataResponseFactory:
|
||||
MAGIC_BYTE_HTTP_REQUEST = "A"
|
||||
|
||||
@staticmethod
|
||||
def create_async_response_message(_id: SnowflakeId) -> DataResponse:
|
||||
@@ -20,9 +20,7 @@ class DataResponseFactory:
|
||||
:param _id: ID
|
||||
:return: Response message
|
||||
"""
|
||||
return DataResponse(
|
||||
str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {}
|
||||
)
|
||||
return DataResponse(str(_id), 200, "application/text", "OK".encode("utf-8"), "", "", {})
|
||||
|
||||
@staticmethod
|
||||
def of(
|
||||
@@ -76,19 +74,19 @@ class DataResponseFactory:
|
||||
:param response: Object to serialize
|
||||
:return: byte array as serialized message
|
||||
"""
|
||||
id_bytes = response.id.encode("utf-8")
|
||||
id_bytes = response.id().encode("utf-8")
|
||||
prolog = (
|
||||
f"|r={response.response_code}|c={response.content_type}|e={response.error}|f={response.error_cause}|t={{{map_as_string(response.trace_info)}}}|b="
|
||||
f"|r={response.response_code()}|c={response.content_type()}|e={response.error()}|f={response.error_cause()}|t={{{map_as_string(response.trace_info())}}}|b="
|
||||
).encode("utf-8")
|
||||
header_length = 1 + len(id_bytes) + len(prolog)
|
||||
prolog_len = header_length.to_bytes(2, byteorder="big")
|
||||
msg_length = 2 + header_length + len(response.response)
|
||||
msg_length = 2 + header_length + len(response.base64body())
|
||||
with BytesIO(bytearray(msg_length)) as baos:
|
||||
baos.write(prolog_len)
|
||||
baos.write(DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST.encode("utf-8"))
|
||||
baos.write(DataMessageMagicByte.HTTP_REQUEST.value.encode("utf-8"))
|
||||
baos.write(id_bytes)
|
||||
baos.write(prolog)
|
||||
baos.write(base64.b64encode(response.response))
|
||||
baos.write(base64.b64encode(response.base64body()))
|
||||
return baos.getvalue()
|
||||
|
||||
@staticmethod
|
||||
@@ -102,7 +100,7 @@ class DataResponseFactory:
|
||||
metadata = input_bytes[2 : prolog_len + 2].decode()
|
||||
|
||||
pattern = re.compile(
|
||||
f"{DataResponseFactory.MAGIC_BYTE_HTTP_REQUEST}"
|
||||
f"{DataMessageMagicByte.HTTP_REQUEST.value}"
|
||||
r"([0-9A-Fa-f]+)\|"
|
||||
r"r=([^|]*)\|"
|
||||
r"c=([^|]*)\|"
|
||||
|
||||
@@ -191,9 +191,7 @@ class StreamingFileHandler(FileHandler):
|
||||
_eof = IN_PROGRESS
|
||||
channel: AbstractRobustChannel = await connection.channel()
|
||||
queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False)
|
||||
logging_debug(
|
||||
f"Awaiting all file chunks being downloaded for {filename}"
|
||||
)
|
||||
logging_debug(f"Awaiting all file chunks being downloaded for {filename}")
|
||||
async with queue.iterator() as queue_iter:
|
||||
async for message in queue_iter:
|
||||
async with message.process():
|
||||
@@ -236,12 +234,8 @@ class StreamingFileHandler(FileHandler):
|
||||
_res = bytearray()
|
||||
try:
|
||||
channel: AbstractRobustChannel = await _connection.channel()
|
||||
queue: AbstractQueue = await channel.get_queue(
|
||||
queue_name, ensure=False
|
||||
)
|
||||
logging_debug(
|
||||
f"Awaiting all buffer chunks being downloaded for {queue_name}"
|
||||
)
|
||||
queue: AbstractQueue = await channel.get_queue(queue_name, ensure=False)
|
||||
logging_debug(f"Awaiting all buffer chunks being downloaded for {queue_name}")
|
||||
async with queue.iterator() as queue_iter:
|
||||
async for message in queue_iter:
|
||||
async with message.process():
|
||||
@@ -296,22 +290,16 @@ class StreamingFileHandler(FileHandler):
|
||||
files_to_download.append(files_info)
|
||||
|
||||
if not files_to_download:
|
||||
logging_info(
|
||||
" [%s] No files to download in this message.", message.delivery_tag
|
||||
)
|
||||
logging_info(" [%s] No files to download in this message.", message.delivery_tag)
|
||||
# await message.ack()
|
||||
return defaultdict()
|
||||
|
||||
# Create a task for each file download
|
||||
_tasks = []
|
||||
for file_info in files_to_download:
|
||||
logging_info(
|
||||
f" [{message.delivery_tag}] about to create task for {file_info}"
|
||||
)
|
||||
logging_info(f" [{message.delivery_tag}] about to create task for {file_info}")
|
||||
_task = asyncio.create_task(
|
||||
self.download_file(
|
||||
loop, rabbitmq_url, file_info, _message_data, output_dir
|
||||
)
|
||||
self.download_file(loop, rabbitmq_url, file_info, _message_data, output_dir)
|
||||
)
|
||||
_tasks.append(_task)
|
||||
logging_info(
|
||||
@@ -377,9 +365,7 @@ class StreamingFileHandler(FileHandler):
|
||||
"total_size": _file_size,
|
||||
"chunk_size": len(_chunk),
|
||||
"eof": (
|
||||
END_OF_FILE
|
||||
if max_chunk_size + _offset >= _file_size
|
||||
else IN_PROGRESS
|
||||
END_OF_FILE if max_chunk_size + _offset >= _file_size else IN_PROGRESS
|
||||
),
|
||||
},
|
||||
)
|
||||
@@ -392,9 +378,7 @@ class StreamingFileHandler(FileHandler):
|
||||
while not _future.done():
|
||||
# sleep allows coroutine to execute. must wait for completion to ensure all chunks written in right order
|
||||
await asyncio.sleep(0.02)
|
||||
logging_debug(
|
||||
f"Waiting for chunk {_offset}/{_file_size} future= {_future}"
|
||||
)
|
||||
logging_debug(f"Waiting for chunk {_offset}/{_file_size} future= {_future}")
|
||||
|
||||
_offset += len(_chunk)
|
||||
logging_debug(
|
||||
|
||||
@@ -2,12 +2,13 @@
|
||||
This module provides serialization and deserialization functions for Pydantic models that extend BaseModel
|
||||
"""
|
||||
|
||||
import uuid
|
||||
import json
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any
|
||||
from typing import Any, Dict
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from pydantic import BaseModel
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
|
||||
from amqp.adapter.logging_utils import logging_debug
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ Helper functions used to aid with (de)serialization of messages sent over AMQP.
|
||||
|
||||
import struct
|
||||
from io import BytesIO
|
||||
from typing import List, Union, Dict
|
||||
from typing import Dict, List, Union
|
||||
|
||||
|
||||
def long_to_bytes(x: int) -> bytes:
|
||||
|
||||
@@ -98,9 +98,7 @@ class ServiceMessageFactory:
|
||||
i += 1
|
||||
reply_to = input_str_array[i] if i < len(input_str_array) else ""
|
||||
i += 1
|
||||
trace_info = (
|
||||
parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
|
||||
)
|
||||
trace_info = parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
|
||||
i += 1
|
||||
message = input_str_array[i] if i < len(input_str_array) else ""
|
||||
return ServiceMessage(
|
||||
@@ -110,11 +108,7 @@ class ServiceMessageFactory:
|
||||
recipient_name=recipient_name,
|
||||
reply_to=reply_to,
|
||||
trace_info=trace_info,
|
||||
message=(
|
||||
base64.b64decode(message).decode()
|
||||
if payload_type & 0x80
|
||||
else message
|
||||
),
|
||||
message=(base64.b64decode(message).decode() if payload_type & 0x80 else message),
|
||||
)
|
||||
except Exception as e:
|
||||
logging_error(
|
||||
@@ -147,9 +141,7 @@ class ServiceMessageFactory:
|
||||
:return: formatted message type
|
||||
"""
|
||||
num_type = (
|
||||
message.message_type.value
|
||||
if message.message_type
|
||||
else ServiceMessageType.UNKNOWN.value
|
||||
message.message_type.value if message.message_type else ServiceMessageType.UNKNOWN.value
|
||||
) - 1
|
||||
fmt = (num_type & 0x7F) | (message.payload_type & 0x80)
|
||||
return f"{fmt:02X}"
|
||||
|
||||
Reference in New Issue
Block a user