Adapter Module

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.

amqp.adapter.cleverthis_service_adapter.is_likely_json(text)

Quick test to see if should convert the response string to JSON format.

Parameters:

text (str)

Return type:

bool

class amqp.adapter.cleverthis_service_adapter.CleverThisServiceAdapter(amq_configuration, routes, session=None, authenticated_user_arg_name='authenticated_user', extra_init_data=None)

Bases: object

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.

Parameters:

• amq_configuration (AMQConfiguration)

• routes (List[APIRoute])

• session (ClientSession)

• authenticated_user_arg_name (str)

__init__(amq_configuration, routes, session=None, authenticated_user_arg_name='authenticated_user', extra_init_data=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.

Parameters:

• amq_configuration (AMQConfiguration)

• routes (List[APIRoute])

• session (ClientSession)

• authenticated_user_arg_name (str)

get_endpoint(endpoint)

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.

Parameters:

endpoint (Any) – The original endpoint to be called.

Returns:

The actual endpoint to be called.

Return type:

Any

clone_and_adapt_route(original_route)

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.

Parameters:

original_route (APIRoute)

Return type:

APIRoute

group_routes_by_method(api_routes)

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).

Parameters:

api_routes (List[APIRoute])

Return type:

Dict[str, List[APIRoute]]

get_content_type(message_headers)

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.

Parameters:

message_headers (Dict[str, List[str]])

Return type:

str

async get_current_user(token)

To Be Overriden in subclass for given CleverThis Service, to return the current_availability user based on the token, in format appropriate for the service.

Parameters:

token (str) – The token to be used for authentication.

Returns:

A user object or None if the user is not authenticated.

Return type:

object

async on_message(message)

Called when a message is received from the AMQP. :param message: :return: AMQP response class with operation status code and optional error details.

Parameters:

message (AMQMessage)

Return type:

DataResponse

async get(message, auth_user)

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.

Parameters:

• message (AMQMessage) – message from the AMQP that contains the GET request

• auth_user (Any) – user authenticated by the API Gateway, as UserSchem

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async put(message, auth_user)

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.

Parameters:

• message (AMQMessage) – message from the AMQP that contains the PUT request

• auth_user (Any) – user authenticated by the API Gateway, as UserSchem

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async post(message, auth_user)

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.

Parameters:

• message (AMQMessage) – message from AMQP that contains the POST request, including body and headers

• auth_user (Any) – user authenticated by the API Gateway, as UserSchem

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async head(message, auth_user)

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.

Parameters:

• message (AMQMessage) – message from AMQP containing the HEAD request, including body and headers

• auth_user (Any) – user authenticated by the API Gateway, as UserSchema

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async delete(message, auth_user)

Handles the DELETE requests for the CleverThis API.

Parameters:

• message (AMQMessage) – message from AMQP that contains the DELETE request, including body & headers

• auth_user (Any) – user authenticated by the API Gateway, as UserSchema

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async options(message, auth_user)

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.

Parameters:

• message (AMQMessage) – message from AMQP that contains the OPTIONS request, including headers

• auth_user (Any) – user authenticated by the API Gateway, as UserSchema

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async patch(message, auth_user)

Handles the PATCH requests for the CleverBRAG API.

Parameters:

• message (AMQMessage) – message from AMQP that contains the PATCH request, including body & headers

• auth_user (Any) – user authenticated by the API Gateway, as UserSchema

Returns:

DataResponse to be sent back to the API Gateway via AMQP

Return type:

DataResponse

async handle_no_body_payload(message, auth_user, method)

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

Parameters:

• message (AMQMessage)

• auth_user (Any)

• method (str)

Return type:

DataResponse

async handle_body_payload(message, auth_user, method)

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

Parameters:

• message (AMQMessage)

• auth_user (Any)

• method (str)

Return type:

DataResponse

async call_cleverthis_api(message, args, api_method, success_code)

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.

Parameters:

• message (AMQMessage) – DataMessage containing the details of the call and the payload

• args (Any) – map of the arguments to pass to the API method

• api_method (Callable[[Any, Any], Coroutine]) – Callable - the method to invoke. Last parameter is the authenticated user

• success_code (int) – HTTP code to be returned when operation suceeds

Returns:

DataResponse class

Return type:

DataResponse

class amqp.adapter.amq_route_factory.AMQRouteFactory

Bases: object

Define the methods to create and validate AMQRoute objects.

static from_string(data)

Builds the AMQRoute from its string form. :param data: route string :return: AMQRoute object

static validate(route_str)

Validates the route string. :param route_str: route string :return: boolean validity indicator

class amqp.adapter.backpressure_handler.ScaleRequestV1(serviceId, taskId, max_availability, current_availability, requestType)

Bases: object

Parameters:

• serviceId (str)

• taskId (str)

• max_availability (int)

• current_availability (int)

• requestType (ScalingRequestAlert)

__init__(serviceId, taskId, max_availability, current_availability, requestType)

Parameters:

• serviceId (str)

• taskId (str)

• max_availability (int)

• current_availability (int)

• requestType (ScalingRequestAlert)

to_json()

Converts the object to a JSON string matching the Java structure

Return type:

str

class amqp.adapter.backpressure_handler.BackpressureHandler(channel, loop, config)

Bases: object

Parameters:

• channel (AbstractRobustChannel)

• loop (AbstractEventLoop)

• config (AMQConfiguration)

last_backpressure_event_time = 0

last_backpressure_event = 'UPDATE'

__init__(channel, loop, config)

Parameters:

• channel (AbstractRobustChannel)

• loop (AbstractEventLoop)

• config (AMQConfiguration)

current_availability = 0

increase_current_load()

Increase the number of parallel executions, or current_availability load

decrease_current_load()

Decrease the number of parallel executions, or current load

update_current_availability(count)

Update the number of parallel executions, or current load

Parameters:

count (int)

update_last_backpressure_event_time()

Update the last data message time

async update_backpressure_value(current_availability, maximum)

Update the current backpressure value and check for overload conditions. This method is called by the AMQService.backpressure() method to update the current parallel executions count and check for overload conditions.

Parameters:

• current_availability (int) – Current value of the backpressure metric

• maximum (int) – Maximum value of the backpressure metric

start_backpressure_monitor()

Return type:

Thread | None

async check_overload_condition()

Check if the current_availability availability is too low (OVERLOAD) or if the service has been idle for too long with high availability (IDLE).

Note: current_availability represents available capacity, not used capacity. - Low availability (close to 0) means OVERLOAD - High availability (close to maximum) with no activity means IDLE

backpressure_monitor_loop()

async handle_backpressure_overload_event()

async publish_backpressure_request(scaling_request)

Parameters:

scaling_request (ScaleRequestV1)

amqp.adapter.consul_global_id_generator.get_config_values(config)

Get configuration values from AMQConfiguration or environment variables.

Returns:

(consul_host, consul_port, consul_counter_key, max_retries, retry_delay_seconds)

Return type:

tuple

Parameters:

config (AMQAdapter)

amqp.adapter.consul_global_id_generator.generate_fallback_id()

Generate a fallback ID when Consul is not available. Uses a combination of timestamp, hostname hash, and random number.

Returns:

A reasonably unique integer ID

Return type:

int

amqp.adapter.consul_global_id_generator.get_unique_instance_id(config)

Get a globally unique ID from Consul. Uses Consul’s atomic Compare-And-Set operations to safely increment a counter. Falls back to a local generation method if Consul is unavailable.

Returns:

A globally unique integer ID

Return type:

int

Parameters:

config (AMQAdapter)

methods to create, serialize and deserialize DataMessage objects.

class amqp.adapter.data_message_factory.DataMessageFactory(machine_id)

Bases: object

Parameters:

machine_id (int)

SNOWFLAKE_ID_BITS = 80

SEQUENCE_BITS = 12

MACHINE_ID_BITS = 24

TIMESTAMP_BITS = 44

MACHINE_ID_MASK = 16777215

SEQUENCE_MASK = 4095

snowflake_epoch = 1727740800.0

__init__(machine_id)

Parameters:

machine_id (int)

classmethod get_instance(machine_id)

return reusable factory instance :param machine_id: a generator ID, should be unique for each component to prevent duplicate IDs :return: factory instance

Parameters:

machine_id (int)

generate_snowflake_id()

Generate an unique ID following the SnowflakeID structure, see https://en.wikipedia.org/wiki/Snowflake_ID :return: 80-bit wide ID

Return type:

SnowflakeId

create_request_message(method, domain, path, headers, message)

Encapsulate / hide the system level inputs, create the AMQ message from business relevant input :param method: a method requested (e.g. HTTP method like GET, POST in case of REST request) :param domain: service host name :param path: path at the service :param headers: HTTP headers :param message: payload :return: DataMessage object

Parameters:

• method (str)

• domain (str)

• path (str)

• headers (Dict[str, List[str]])

Return type:

DataMessage

safe_rsplit(fq_method_name, default='_')

Safely split a fully qualified method name into package, class, and method

Parameters:

• fq_method_name (str) – Fully qualified method name

• default (str, optional) – Default value for missing tokens. Defaults to ‘’.

Returns:

(package, class_name, method)

Return type:

tuple

create_rpc_message(fq_method_name, swarm_id, args=None)

Create a RPC message with the given parameters. :param fq_method_name: fully qualified method name (e.g. ‘com.example.ServiceClass.method_name’) :param args: arguments whose serialized form will form the request body :return: DataMessage object

Parameters:

• fq_method_name (str)

• swarm_id (str)

• args (Dict[str, Any])

Return type:

DataMessage

static amq_message_routing_key(message)

Constructs the message specific routing key. :param message: DataMessage input :return: routing key (as string) compliant to RabbitMQ allowed character set.

Parameters:

message (DataMessage)

Return type:

str

static get_timestamp_from_id(_id)

Helper method - retrieve the timestamp from the ID :param _id: SnowflakeID (includes encoded timestamp) :return: the message timestamp

Parameters:

_id (SnowflakeId)

Return type:

int

static to_string(message)

Human-readable representation. :param message: DataMessage input :return: as string

Parameters:

message (DataMessage)

Return type:

str

static serialize(message)

Converts DataMessage to byte array :param message: message to serialize :return: messages as bytes

Parameters:

message (DataMessage)

Return type:

bytes

static from_bytes(input_bytes)

Builds the DataMessage from its serialized (byte array) form. :param input_bytes: serialized message :return: DataMessage instance

Parameters:

input_bytes (bytes)

Return type:

DataMessage

async static from_stream(stream, rabbitmq_url, loop, file_handler=<amqp.adapter.file_handler.StreamingFileHandler object>)

Builds the DataMessage from its serialized (byte array) form. :param stream: serialized message :param rabbitmq_url: RabbitMQ connection URL :param loop: event loop :param file_handler: file handler for downloading the message :return: DataMessage instance

Parameters:

• stream (bytes)

• rabbitmq_url (str)

• file_handler (FileHandler)

Return type:

DataMessage | None

class amqp.adapter.data_parser.MultipartFormDataParser(message)

Bases: object

Parses multipart/form-data content and extracts form fields (key/value pairs) and files. Note: multipart message that contains file(s) to upload are converted to CleverMicro JSON-based streaming format, where files are transmitted separately via stream using dedicated queue, so the message content is no longer the HTTP multipart format, but the Content-Type is being preserved for compatibility with the original message. This class supports both formats (multipart and JSON) depending on the message content, because if no file is included, the message is transmitted in the unchanged multipart/form-data format.

Parameters:

message (AMQMessage)

__init__(message)

Parameters:

message (AMQMessage)

on_field(field)

Parameters:

field (Field)

on_file(file)

Parameters:

file (File)

class amqp.adapter.data_parser.AMQDataParser

Bases: object

A class for parsing AMQ data messages. It provides methods to parse form data,

static process_form_data(json_input)

Processes the input JSON to extract form data. The JSON has two top entries: ‘files’ and ‘form’. This is part of the file stream implementation in AMQ adapter to stream large files in chunks. ‘files’ contains information about the AMQ queue from which read the file, while ‘form’ contains additional non-file form values. This function combines the file and values at the top level in the returned dictionary. :param json_input: JSON string or dict containing the form and files data

Returns:

Processed form data with files data overriding non-empty values

Return type:

dict

static parse_get_url(url)

Parses an HTTP GET URL and returns (context_path, query_params_dict)

Parameters:

url (str) – The URL string to parse (e.g., “http://example.com/path?foo=1&bar=2&foo=3”)

Returns:

(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

Return type:

tuple

static parse_request_body(message)

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. :param message: Contains raw request and MIME headers. :type message: DataMessage

Returns:

Parsed data as a dictionary.

Return type:

dict

Parameters:

message (AMQMessage)

method to create, serialize and deserialize DataResponse message

class amqp.adapter.data_response_factory.DataResponseFactory

Bases: object

static create_async_response_message(_id)

Helper method to create standard OK response. :param _id: ID :return: Response message

Parameters:

_id (SnowflakeId)

Return type:

DataResponse

static of(id, response_code, content_type, body, trace_info, error=None, error_cause=None)

Create the DataResponse message supplying all key values :param id: SnowflakeID :param response_code: HTTP response code :param content_type: MIME content type :param body: payload :param trace_info: Jaeger trace info (id) :return: DataResponseMessage object

Parameters:

• id (SnowflakeId)

• response_code (int)

• content_type (str)

• body (bytes)

• trace_info (dict[str, str])

• error (str | None)

• error_cause (str | None)

Return type:

DataResponse

static of_error_message(id, error_message, trace_info)

Parameters:

• id (SnowflakeId)

• error_message (AMQErrorMessage)

• trace_info (dict[str, str])

Return type:

DataResponse

static serialize(response)

Serialize the message to byte array (to be sent over RabbitMQ) :param response: Object to serialize :return: byte array as serialized message

Parameters:

response (DataResponse)

Return type:

bytes

static from_bytes(input_bytes)

Buil;ds the DataResponse object from its serialized form :param input_bytes: byte array - serialized response :return: DataResponse object

Parameters:

input_bytes (bytes)

Return type:

DataResponse

amqp.adapter.data_response_factory.AMQResponseFactory

alias of DataResponseFactory

implements the file downloader for the amqp adapter, where file is sent in chunks over RabbitMQ

class amqp.adapter.file_handler.FileHandler

Bases: object

The abstract base class for file handling. This is mainly to make the code testable, as there are only 2 options for pushing (uploading) a file(s) to a service: as HTTP multipart message delivered via RabbitMQ or via stream in RabbitMQ. Downloading a file is always done via RabbitMQ stream. Thus the method signiture is tighly coupled to the RabbitMQ API in v1. Only alternative implementation is a mock object in unittests.

__init__()

ensure_directory_exists(filepath)

Ensures that the directory for the given filepath exists, and handles file versioning.

Parameters:

filepath – The path to the file.

Returns:

The original filepath if the file does not exist, or a modified filepath pointing to the next available version of the file if it does.

Return type:

str

async download_buffer(loop, rabbitmq_url, queue_name)

Asynchronously downloads a bytearray from RabbitMQ based on the provided file information. It assumes entire array is in the first (and only) chunk :param loop: The asyncio event loop. :param rabbitmq_url: The RabbitMQ connection URL. :param queue_name: The directory where the file should be saved. :type queue_name: str

Return type:

(<class ‘bytearray’>, <class ‘int’>)

async download_file(loop, rabbitmq_url, file_info, message_data, output_dir)

Asynchronously downloads a file from RabbitMQ based on the provided file information. takes the file_info and message_data and output_dir, creates queue consumer and waits for the file to be downloaded. File may consist of multiple chunks, so the function will wait until all chunks are received. :param loop: The asyncio event loop. :param rabbitmq_url: The RabbitMQ connection URL. :param file_info: A dictionary containing file metadata :type file_info: dict :param message_data: The entire message :type message_data: dict :param output_dir: The directory where the file should be saved. :type output_dir: str

Return type:

(<class ‘int’>, <class ‘int’>)

async on_message(message, json_data, rabbitmq_url, loop, output_dir='downloaded_files')

Asynchronous callback function for handling messages from the ‘amq-cleverswarm’ queue.

Parameters:

• message (AbstractIncomingMessage) – The incoming RabbitMQ message.

• rabbitmq_url (str) – The RabbitMQ connection URL.

• json_data (bytes)

Return type:

defaultdict

async publish_file(exchange, file_path, routing_key, loop, max_chunk_size=1048576, content_type='application/octet-stream', delivery_mode=DeliveryMode.PERSISTENT)

Reads a file from the file system and publishes its content to a RabbitMQ exchange in chunks. It declares a queue with a random name, binds it to the specified exchange, and uses the queue name as the routing key.

Parameters:

• exchange (AbstractExchange) – The aio_pika RobustChannel to use for communication with RabbitMQ.

• file_path (Path) – The path to the file to publish.

• max_chunk_size (int) – The maximum size of each chunk (in bytes).

• routing_key (str) – The name of the RabbitMQ exchange routing key to publish to correct queue.

• loop (AbstractEventLoop) – correct event loop

• max_chunk_size – chunk size to send

• content_type (str) – The content type of the file data.

• delivery_mode (DeliveryMode) – The delivery mode (e.g., PERSISTENT or TRANSIENT).

Returns:

The name of the randomly generated queue where the file content was published.

Return type:

tuple[str, int]

class amqp.adapter.file_handler.StreamingFileHandler

Bases: FileHandler

Handles file downloads from RabbitMQ streams.

__init__()

async download_file(loop, rabbitmq_url, file_info, message_data, output_dir)

Asynchronously downloads a file from RabbitMQ based on the provided file information. takes the file_info and message_data and output_dir, creates queue consumer and waits for the file to be downloaded :param loop: The asyncio event loop. :param rabbitmq_url: The RabbitMQ connection URL. :param file_info: A dictionary containing file metadata :type file_info: dict :param message_data: The entire message :type message_data: dict :param output_dir: The directory where the file should be saved. :type output_dir: str

Return type:

(<class ‘int’>, <class ‘int’>)

async download_buffer(loop, rabbitmq_url, queue_name)

Asynchronously downloads a bytearay from RabbitMQ based on the provided file information. It assumes entire array is in the first (and only) chunk :param loop: The asyncio event loop. :param rabbitmq_url: The RabbitMQ connection URL. :param queue_name: The directory where the file should be saved. :type queue_name: str

Parameters:

queue_name (str | None)

Return type:

(<class ‘bytearray’>, <class ‘int’>)

async on_message(message, json_data, rabbitmq_url, loop, output_dir='downloaded_files')

Asynchronous callback function for handling messages from the ‘amq-cleverswarm’ queue.

Parameters:

• message (AbstractIncomingMessage) – The incoming RabbitMQ message.

• json_data (bytes) – The JSON data from the message.

• rabbitmq_url (str) – The RabbitMQ connection URL.

• loop – The asyncio event loop of the RabbitMQ API

• output_dir

Return type:

defaultdict

async publish_file(exchange, file_path, routing_key, loop, max_chunk_size=1048576, content_type='application/octet-stream', delivery_mode=DeliveryMode.PERSISTENT)

Reads a file from the file system and publishes its content to a RabbitMQ exchange in chunks. It declares a queue with a random name, binds it to the specified exchange, and uses the queue name as the routing key.

Parameters:

• exchange (AbstractExchange) – The aio_pika RobustChannel to use for communication with RabbitMQ.

• file_path (Path) – The path to the file to publish.

• max_chunk_size (int) – The maximum size of each chunk (in bytes).

• routing_key (str) – The name of the RabbitMQ exchange routing key to publish to correct queue.

• loop (AbstractEventLoop) – correct event loop

• max_chunk_size – chunk size to send

• content_type (str) – The content type of the file data.

• delivery_mode (DeliveryMode) – The delivery mode (e.g., PERSISTENT or TRANSIENT).

Returns:

The name of the randomly generated queue where the file content was published.

Return type:

tuple[str, int]

amqp.adapter.logging_utils.get_context_info()

Retrieves the current_availability module, line number, and function name.

Returns:

A tuple containing (module name, line number, function name). Returns (None, None, None) if it fails to retrieve the information.

amqp.adapter.logging_utils.logging_error(msg, *args)

Log an error message according to current_availability logging for ‘ERROR’ level. Add module name, line and function name where the error occurred, on single line. :param msg: The error message to log. :type msg: str

Parameters:

msg (str)

Return type:

None

amqp.adapter.logging_utils.logging_warning(msg, *args)

Log a warning message according to current_availability logging for ‘WARN’ level. Add module name, line and function name where the print occurred, on single line. :param msg: The warning message to log. :type msg: str

Parameters:

msg (str)

Return type:

None

amqp.adapter.logging_utils.logging_info(msg, *args)

Log an info message according to current_availability logging for ‘INFO’ level. Add module name, line and function name where the print occurred, on single line. :param msg: The info message to log. :type msg: str

Parameters:

msg (str)

Return type:

None

amqp.adapter.logging_utils.logging_debug(msg, *args)

Log a debug message according to current_availability logging for ‘DEBUG’ level. Add module name, line and function name where the print occurred, on single line. :param msg: The debug message to log. :type msg: str

Parameters:

msg (str)

Return type:

None

This module provides serialization and deserialization functions for Pydantic models that extend BaseModel

amqp.adapter.pydantic_serializer.serialize_object(obj)

Serializes an R2RSerializable object to a string in the format “<ClassName>:{key1:value1, key2:value2,…}”. Handles UUIDs and datetimes, and nested Pydantic models.

Parameters:

obj (BaseModel) – The BaseModel object to serialize.

Returns:

A string representation of the object.

Return type:

str

amqp.adapter.pydantic_serializer.python_type_to_json(data)

Convert Python objects (dict, list, tuple, set, datetime, etc.) to a JSON string. Handles common non-JSON-serializable types like datetime, date, and sets.

Parameters:

data – Python object to be converted (dict, list, tuple, set, etc.)

Returns:

JSON string representation of the input data

Return type:

str

Raises:

TypeError – If the object contains non-serializable types (e.g., custom classes)

amqp.adapter.pydantic_serializer.try_deserialize_object(data)

Try deserialize object, if failed, return the input string as-is.

Parameters:

data (str)

Return type:

BaseModel | str

amqp.adapter.pydantic_serializer.deserialize_object(data)

Deserializes a string in the format “<ClassName>:{key1:value1, key2:value2,…}” back into an BaseModel object. Handles UUIDs, datetimes, and nested Pydantic models.

Parameters:

data (str) – The string to deserialize.

Returns:

An R2RSerializable object.

Return type:

BaseModel

amqp.adapter.pydantic_serializer.parse_url_query(url)

Parses a HTTP GET URL and returns (context_path, query_params_dict)

Parameters:

url – The URL string to parse (e.g., “http://example.com/path?foo=1&bar=2&foo=3”)

Returns:

(context_path, query_params_dict) - context_path: The path part of the URL (e.g., “/path”) - query_params_dict: Dictionary of query parameters where values are always lists

(e.g., {“foo”: [“1”, “3”], “bar”: [“2”]})

Return type:

tuple

Helper functions used to aid with (de)serialization of messages sent over AMQP.

amqp.adapter.serializer.long_to_bytes(x)

Parameters:

x (int)

Return type:

bytes

amqp.adapter.serializer.bytes_to_long(b)

Parameters:

b (bytes)

Return type:

int

amqp.adapter.serializer.read_long(bis)

Parameters:

bis (BytesIO)

Return type:

int

amqp.adapter.serializer.object_or_list_as_string(value)

Parameters:

value (str | List[str])

Return type:

str

amqp.adapter.serializer.map_as_string(map_)

Parameters:

map_ (Dict[str, str])

Return type:

str

amqp.adapter.serializer.map_with_list_as_string(map_)

Parameters:

map_ (Dict[str, List[str]])

Return type:

str

amqp.adapter.serializer.parse_map_string(input_str)

Parameters:

input_str (str)

Return type:

Dict[str, str]

amqp.adapter.serializer.csv_as_list(input_str)

Parameters:

input_str (str)

Return type:

List[str]

amqp.adapter.serializer.add_to_map(map_, token)

Parameters:

• map_ (Dict[str, List[str]])

• token (str)

Return type:

None

amqp.adapter.serializer.parse_map_list_string(input_str)

Parameters:

input_str (str)

Return type:

Dict[str, List[str]]

amqp.adapter.serializer.str_to_bool(value)

Parameters:

value (str)

Return type:

bool

class amqp.adapter.service_message_factory.ServiceMessageFactory(name)

Bases: object

build/serialize/deserialize CleverMicro Service Message class

Parameters:

name (str)

__init__(name)

Parameters:

name (str)

to_string(message)

Convert ServiceMessage to human-readable format :param message: ServiceMessage object :return: as string

Parameters:

message (ServiceMessage)

Return type:

str

of(message_type, payload, recipient_name)

Builds ServiceMessage object from relevant values. :param message_type: enum Message type :param payload: Message content, stored as-is :param recipient_name: Recipient name (from the routing expression) :return: ServiceMessage object

Parameters:

• message_type (ServiceMessageType)

• payload (str)

• recipient_name (str)

Return type:

ServiceMessage

as_base64(message_type, payload, recipient_name)

Builds ServiceMessage object from relevant values, stores content Base64 encoded. :param message_type: enum Message type :param payload: Message content, stored as Base64 string :param recipient_name: Recipient name (from the routing expression) :return: ServiceMessage object

Parameters:

• message_type (ServiceMessageType)

• payload (str)

• recipient_name (str)

Return type:

ServiceMessage

from_bytes(input_bytes)

Deserialize ServiceMessage from byte array serialized form. :param input_bytes: serialized ServiceMessage :return: ServiceMessage object represented by the input.

Parameters:

input_bytes (bytes)

Return type:

ServiceMessage

serialize(message)

Serialize ServiceMessage to byte array. :param message: ServiceMessage to serialize. :return: serialized object.

Parameters:

message (ServiceMessage)

Return type:

bytes

format_message_type(message)

Formats the Enum message type and decorates it with base64 flag at highest bit. For compatibility with 0-based enum ordinals in Java subtract 1 from ordinal value. :param message: Service message to send :return: formatted message type

Parameters:

message (ServiceMessage)

Return type:

str

class amqp.adapter.trace_info_adapter.TraceInfoAdapter

Bases: object

TODO This class should set up the Open Telemetry tracing stack. To be done later.

static create_produce_span(message_id)

Parameters:

message_id (str)

Return type:

Dict[str, str]

__init__()

Return type:

None

class amqp.adapter.type_descriptor.TypeDescriptor(type_string, extra_init_data)

Bases: object

Parameters:

• type_string (str)

• extra_init_data (dict)

__init__(type_string, extra_init_data)

Parameters:

• type_string (str)

• extra_init_data (dict)

instantiate(payload, is_json=False)

Parameters:

• payload (Any)

• is_json (bool)

Return type:

Any

amqp.adapter.type_descriptor.merge_extra_init_data(cls, json_payload, extra_init_data)

Merges extra_init_data into json_payload only when: 1. cls has an attribute named like the key 2. json_payload doesn’t already have the key

Parameters:

extra_init_data (dict)