minos.networks.brokers.messages module

class minos.networks.brokers.messages.BrokerMessage(topic, data, *, identifier=None, status=None, strategy=None, headers=None, **kwargs)

Bases: minos.common.model.declarative.DeclarativeModel

Broker Message class.

__init__(topic, data, *, identifier=None, status=None, strategy=None, headers=None, **kwargs)

Class constructor.

Parameters

kwargs – Named arguments to be set as model attributes.

property avro_bytes: bytes

Generate bytes representation of the current instance.

Return type

bytes

Returns

A bytes object.

property avro_data: dict[str, typing.Any]

Compute the avro data of the model.

Return type

dict[str, typing.Any]

Returns

A dictionary object.

avro_schema = [{'name': 'BrokerMessage', 'namespace': 'minos.networks.brokers.messages.e99f607f-f846-4379-9e9b-075a7f73dab6', 'type': 'record', 'fields': [{'name': 'topic', 'type': 'string'}, {'name': 'data', 'type': 'null'}, {'name': 'identifier', 'type': {'type': 'string', 'logicalType': 'uuid'}}, {'name': 'reply_topic', 'type': ['string', 'null']}, {'name': 'user', 'type': [{'type': 'string', 'logicalType': 'uuid'}, 'null']}, {'name': 'status', 'type': 'int'}, {'name': 'strategy', 'type': 'string'}, {'name': 'headers', 'type': {'type': 'map', 'values': 'string'}}]}]

property avro_str: str

Generate bytes representation of the current instance.

Return type

str

Returns

A bytes object.

classname = 'minos.networks.brokers.messages.BrokerMessage'

data: Any

property fields: dict[str, minos.common.model.fields.Field]

Fields getter

Return type

dict[str, minos.common.model.fields.Field]

classmethod from_avro(schema, data)

Build a new instance from the avro schema and data.

Parameters

• schema (typing.Union[dict[str, typing.Any], list[dict[str, typing.Any]]]) – The avro schema of the model.

• data (dict[str, typing.Any]) – The avro data of the model.

Return type

~T

Returns

A new DynamicModel instance.

classmethod from_avro_bytes(raw, **kwargs)

Build a single instance or a sequence of instances from bytes

Parameters

raw (bytes) – A bytes data.

Return type

typing.Union[~T, list[~T]]

Returns

A single instance or a sequence of instances.

classmethod from_avro_str(raw, **kwargs)

Build a single instance or a sequence of instances from bytes

Parameters

raw (str) – A bytes data.

Return type

typing.Union[~T, list[~T]]

Returns

A single instance or a sequence of instances.

classmethod from_model_type(model_type, *args, **kwargs)

Build a DeclarativeModel from a ModelType.

Parameters

• model_type (minos.common.model.types.model_types.ModelType) – ModelType object containing the model structure.

• args – Positional arguments to be passed to the model constructor.

• kwargs – Named arguments to be passed to the model constructor.

Return type

~T

Returns

A new DeclarativeModel instance.

classmethod from_typed_dict(typed_dict, *args, **kwargs)

Build a Model from a TypeDict and data.

Parameters

• typed_dict (typing.TypedDict) – TypeDict object containing the DTO’s structure

• args – Positional arguments to be passed to the model constructor.

• kwargs – Named arguments to be passed to the model constructor.

Return type

~T

Returns

A new DataTransferObject instance.

get(k[, d]) → D[k] if k in D, else d.  d defaults to None.

headers: dict[str, str]

identifier: UUID

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

property ok: bool

Check if the reply is okay or not.

Return type

bool

Returns

True if the reply is okay or False otherwise.

reply_topic: Optional[str]

status: BrokerMessageStatus

strategy: BrokerMessageStrategy

classmethod to_avro_bytes(models)

Create a bytes representation of the given object instances.

Parameters

models (list[~T]) – A sequence of minos models.

Return type

bytes

Returns

A bytes object.

classmethod to_avro_str(models)

Create a bytes representation of the given object instances.

Parameters

models (list[~T]) – A sequence of minos models.

Return type

str

Returns

A bytes object.

topic: str

type_hints

type_hints_parameters = ()

user: Optional[UUID]

values() → an object providing a view on D's values

class minos.networks.brokers.messages.BrokerMessageStatus(value)

Bases: enum.IntEnum

Broker Message Status class.

ERROR = 400

SUCCESS = 200

SYSTEM_ERROR = 500

class minos.networks.brokers.messages.BrokerMessageStrategy(value)

Bases: str, enum.Enum

Broker Message Strategy class

MULTICAST = 'multicast'

UNICAST = 'unicast'