Model Module

amqp.model.model.RS = ':': Define the structure of messages used in Clever Micro

class amqp.model.model.DataMessageMagicByte(*values)[source]

Bases: Enum

HTTP_REQUEST = 'A'

RPC_REQUEST = 'R'

class amqp.model.model.CleverMicroMessage(magic: str, id: amqp.model.snowflake_id.SnowflakeId, m_field: str, d_field: str, p_field: str, headers: Dict[str, List[str]], trace_info: Dict[str, str], base64body: bytes)[source]

Bases: object

Parameters:

magic (str)
id (SnowflakeId)
m_field (str)
d_field (str)
p_field (str)
headers (Dict[str, List[str]])
trace_info (Dict[str, str])
base64body (bytes)

DEFAULT_CONTENT_TYPE = ['application/octet-stream']

The common structure of a message that transfers the user request to the Service and back. All data related messages conform to this base structure:

magic - identifies the message type (REST, RPC, DataResponse) id - unique message ID m_field - method name (for REST style calls) or service name (for RPC style calls) or response code for DataResponse d_field - domain name (for REST style calls) or module name (for RPC style calls) or error code for DataResponse p_field - path (for REST style calls) or method name (for RPC style calls) or error cause for DataResponse headers - headers of the message (e.g. Content-Type, Accept, etc.) trace_info - trace information (e.g. trace ID, span ID, etc. for Jaeger trace monitor) base64body - the body of the message encoded in base64

__init__(magic, id, m_field, d_field, p_field, headers, trace_info, base64body)[source]

Parameters:

magic (str)
id (SnowflakeId)
m_field (str)
d_field (str)
p_field (str)
headers (Dict[str, List[str]])
trace_info (Dict[str, str])
base64body (bytes)

magic()[source]

id()[source]

m_field()[source]

Return type:: str

d_field()[source]

Return type:: str

p_field()[source]

Return type:: str

headers()[source]

Return type:: Dict[str, List[str]]

trace_info()[source]

Return type:: Dict[str, str]

base64body()[source]

Return type:: bytes

body()[source]

Return type:: bytes

content_type()[source]

Return type:: str

return_address()[source]

Return type:: str

class amqp.model.model.DataMessage(magic, id, method, domain, path, headers, trace_info, base64body)[source]

Bases: CleverMicroMessage

The class represents a structure of a message that transfers the user request to the Service. It may also be used to make a call/request between services. The response to this call is in format of DataResponse class.

See DataMessageFactory for code to instantiate, serialize and deserialize the class

Parameters:

magic (str)
id (SnowflakeId)
method (str)
domain (str)
path (str)
headers (Dict[str, List[str]])
trace_info (Dict[str, str])
base64body (bytes)

__init__(magic, id, method, domain, path, headers, trace_info, base64body)[source]

Parameters:

magic (str)
id (SnowflakeId)
method (str)
domain (str)
path (str)
headers (Dict[str, List[str]])
trace_info (Dict[str, str])
base64body (bytes)

method()[source]

Return type:: str

domain()[source]

Return type:: str

path()[source]

Return type:: str

routing_key()[source]

Return type:: str

class amqp.model.model.MultipartDataMessage(message, extra_data)[source]

Bases: DataMessage

Parameters:

message (DataMessage)
extra_data (dict)

__init__(message, extra_data)[source]

Parameters:

message (DataMessage)
extra_data (dict)

class amqp.model.model.AMQMessage(data_message, connection)[source]

Bases: 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.

Parameters:

data_message (DataMessage)
connection (AbstractRobustConnection)

__init__(data_message, connection)[source]

Parameters:

data_message (DataMessage)
connection (AbstractRobustConnection)

property connection: AbstractRobustConnection: Get the AMQP connection associated with this message

class amqp.model.model.DataResponse(magic, id, response_code, content_type, error, error_cause, headers, trace_info, base64body)[source]

Bases: CleverMicroMessage

The response message to REST & RPC style calls (note: the request is made with DataMessage)

See DataResponseFactory for code to instantiate, serialize and deserialize the class

Parameters:

magic (str)
id (SnowflakeId)
response_code (str)
content_type (str)
error (str)
error_cause (str)
headers (Dict[str, List[str]])
trace_info (Dict[str, str])
base64body (bytes)

__init__(magic, id, response_code, content_type, error, error_cause, headers, trace_info, base64body)[source]

Parameters:

magic (str)
id (SnowflakeId)
response_code (str)
content_type (str)
error (str)
error_cause (str)
headers (Dict[str, List[str]])
trace_info (Dict[str, str])
base64body (bytes)

response_code()[source]

Return type:: str

error()[source]

Return type:: str

error_cause()[source]

Return type:: str

class amqp.model.model.AMQRoute(component_name, exchange, queue, key, timeout, valid_until)[source]

Bases: object

Configuration item representing one AMQRoute. Adapter uses this to declare the Exchange and/or queue and bind the key as routing expression.

Parameters:

component_name (str) – a service/component name that this route points to (owner of the route)
exchange (str) – RabbitMQ exchange name
queue (str) – RabbitMQ queue name
key (str) – RabbitMQ TOPIC Exchange routing key
timeout (int) – Timeout in milliseconds, how long to wait for reply, or not waiting if less then 1
valid_until (int) – optional timestamp (millis from epoch) after which the route is ignored

component_name: str

exchange: str

queue: str

key: str

timeout: int

valid_until: int

FORMAT = 'Component-Name:Exchange-Name:Queue-Name:Routing-Key:Timeout:Valid-Until'

property primary_key: Build the unique identifier of the route :return primary key

as_string()[source]

Return type:: str

__init__(component_name, exchange, queue, key, timeout, valid_until)

Parameters:

component_name (str)
exchange (str)
queue (str)
key (str)
timeout (int)
valid_until (int)

Return type:

None

amqp.model.model.AMQResponse: alias of DataResponse

class amqp.model.model.ServiceMessage(id=None, payload_type=0, message_type=ServiceMessageType.UNKNOWN, recipient_name=None, reply_to=None, trace_info=<factory>, message=None)[source]

Bases: object

The CleverMicro service message that is used to communicate the internal state of the AMQ channel. In v0.2 it supports service discovery / route registration and Backpressure status notification.

Parameters:

id (SnowflakeId)
payload_type (int)
message_type (ServiceMessageType)
recipient_name (str)
reply_to (str)
trace_info (Dict[str, str])
message (str)

id: SnowflakeId = None

payload_type: int = 0

message_type: ServiceMessageType = 6

recipient_name: str = None

reply_to: str = None

trace_info: Dict[str, str]

message: str = None

is_valid()[source]

Return type:: bool

__init__(id=None, payload_type=0, message_type=ServiceMessageType.UNKNOWN, recipient_name=None, reply_to=None, trace_info=<factory>, message=None)

Parameters:

id (SnowflakeId)
payload_type (int)
message_type (ServiceMessageType)
recipient_name (str)
reply_to (str)
trace_info (Dict[str, str])
message (str)

Return type:

None

class amqp.model.model.ScalingRequestAlert(*values)[source]

Bases: Enum

OVERLOAD = 'OVERLOAD'

IDLE = 'IDLE'

UPDATE = 'UPDATE'

SCALE_UP = 'SCALE_UP'

SCALE_DOWN = 'SCALE_DOWN'

class amqp.model.model.ScalingRequest(service_id: str, task_id: str, max_availability: int, current_availability: int, request_type: amqp.model.model.ScalingRequestAlert, version: int = 1)[source]

Bases: object

Parameters:

service_id (str)
task_id (str)
max_availability (int)
current_availability (int)
request_type (ScalingRequestAlert)
version (int)

service_id: str

task_id: str

max_availability: int

current_availability: int

request_type: ScalingRequestAlert

version: int = 1

to_json()[source]

Converts the object to a JSON string matching the Java structure

Return type:: str

classmethod from_json(json_str)[source]

Creates a ScalingRequest from a JSON string

Parameters:: json_str (str)
Return type:: ScalingRequest

__init__(service_id, task_id, max_availability, current_availability, request_type, version=1)

Parameters:

service_id (str)
task_id (str)
max_availability (int)
current_availability (int)
request_type (ScalingRequestAlert)
version (int)

Return type:

None

class amqp.model.model.AMQErrorMessage(response_code, error, detail, content_type='application/json')[source]

Bases: object

The error response message to RPC style call (note: the request is made with DataMessage)

Parameters:

response_code (int)
error (str)
detail (str)
content_type (str)

__init__(response_code, error, detail, content_type='application/json')[source]

Parameters:

response_code (int)
error (str)
detail (str)
content_type (str)

serialize()[source]

Return type:: bytes

class amqp.model.service_message_type.ServiceMessageType(*values)[source]

Bases: Enum

ROUTING_DATA_REQUEST = 1

ROUTING_DATA = 2

ROUTING_DATA_REMOVE = 3

ROUTING_DATA_REPLACE = 4

BACKPRESSURE = 5

UNKNOWN = 6

class amqp.model.snowflake_id.SnowflakeId(hi_bits_or_bytes=None, lo_bits=None)[source]

Bases: object

see https://en.wikipedia.org/wiki/Snowflake_ID

SNOWFLAKE_ID_WIDTH = 10

__init__(hi_bits_or_bytes=None, lo_bits=None)[source]

static from_hex(_input)[source]

get_bytes()[source]

get_bits()[source]

as_string()[source]

Return type:: str