minos.aggregate.events.models

Classes

Event

Event class.

class Event[source]

Bases: DeclarativeModel

Event class.

uuid: UUID

name: str

version: int

action: Action

created_at: datetime

fields_diff: FieldDiffContainer

property simplified_name: str

Get the RootEntity’s simplified name.

Returns: An string value.

get_one(name, return_diff=False)[source]

Get first field diff with given name.

Parameters

name (str) – The name of the field diff.
return_diff (bool) – If True the result is returned as field diff instances, otherwise the result is returned as value instances.

Returns

A FieldDiff instance.

Return type

Union[FieldDiff, Any, list[minos.aggregate.events.fields.FieldDiff], list[Any]]

get_field(name, return_diff=False)[source]

Get first field diff with given name.

Parameters

name (str) – The name of the field diff.
return_diff (bool) – If True the result is returned as field diff instances, otherwise the result is returned as value instances.

Returns

A FieldDiff instance.

Return type

Union[FieldDiff, Any, list[minos.aggregate.events.fields.FieldDiff], list[Any]]

get_all(return_diff=False)[source]

Get all field diffs with given name.

Parameters: return_diff (bool) – If True the result is returned as field diff instances, otherwise the result is returned as value instances.
Returns: A list of FieldDiff instances.
Return type: dict[str, Union[minos.aggregate.events.fields.FieldDiff, Any, list[minos.aggregate.events.fields.FieldDiff], list[Any]]]

get_fields(return_diff=False)[source]

Get all field diffs with given name.

Parameters: return_diff (bool) – If True the result is returned as field diff instances, otherwise the result is returned as value instances.
Returns: A list of FieldDiff instances.
Return type: dict[str, Union[minos.aggregate.events.fields.FieldDiff, Any, list[minos.aggregate.events.fields.FieldDiff], list[Any]]]

classmethod from_difference(a, b, action=Action.UPDATE)[source]

Build an Event instance from the difference of two instances.

Parameters

a (RootEntity) – One RootEntity instance.
b (RootEntity) – Another RootEntity instance.
action (Action) – The action that generates the RootEntity difference.

Returns

An Event instance.

Return type

Event

classmethod from_root_entity(instance, action=Action.CREATE)[source]

Build an Event from a RootEntity (considering all fields as differences).

Parameters

instance (RootEntity) – A RootEntity instance.
action (Action) – The action that generates the event.

Returns

An Event instance.

Return type

Event

classmethod from_deleted_root_entity(instance, action=Action.DELETE)[source]

Build an Event from a RootEntity (considering all fields as differences).

Parameters

instance (RootEntity) – A RootEntity instance.
action (Action) – The action that generates the event.

Returns

An Event instance.

Return type

Event

decompose()[source]

Decompose the Event fields into multiple Event instances with once Field.

Returns: An list of``Event`` instances.
Return type: list[minos.aggregate.events.models.Event]

__init__(*args, **kwargs)

Class constructor.

Parameters: kwargs – Named arguments to be set as model attributes.

property avro_bytes: bytes

Generate bytes representation of the current instance.

Returns: A bytes object.

property avro_data: dict[str, Any]

Compute the avro data of the model.

Returns: A dictionary object.

avro_schema = [{'name': 'Event', 'namespace': 'minos.aggregate.events.models.3c8467af-0bca-4048-8752-747b43b426ed', 'type': 'record', 'fields': [{'name': 'uuid', 'type': {'type': 'string', 'logicalType': 'uuid'}}, {'name': 'name', 'type': 'string'}, {'name': 'version', 'type': 'int'}, {'name': 'action', 'type': {'type': 'string', 'logicalType': 'minos.aggregate.actions.Action'}}, {'name': 'created_at', 'type': {'type': 'long', 'logicalType': 'timestamp-micros'}}, {'name': 'fields_diff', 'type': {'name': 'FieldDiffContainer', 'namespace': 'minos.aggregate.events.fields.9643fffa-0d4d-45f6-b388-ee2bdbcabae2', 'type': 'record', 'fields': []}}]}]

property avro_str: str

Generate bytes representation of the current instance.

Returns: A bytes object.

classname = 'minos.aggregate.events.models.Event'

static decode_data(decoder, target, type_, **kwargs)

Decode data with the given decoder.

Parameters

decoder (DataDecoder) – The decoder instance.
target (Any) – The data to be decoded.
type – The data type.
kwargs – Additional named arguments.
type_ (ModelType) –

Returns

A decoded instance.

Return type

Any

static decode_schema(decoder, target, **kwargs)

Decode schema with the given encoder.

Parameters

decoder (SchemaDecoder) – The decoder instance.
target (Any) – The schema to be decoded.
kwargs – Additional named arguments.

Returns

The decoded schema as a type.

Return type

Any

static encode_data(encoder, target, **kwargs)

Encode data with the given encoder.

Parameters

encoder (DataEncoder) – The encoder instance.
target (Any) – An optional pre-encoded data.
kwargs – Additional named arguments.

Returns

The encoded data of the instance.

Return type

Any

static encode_schema(encoder, target, **kwargs)

Encode schema with the given encoder.

Parameters

encoder (SchemaEncoder) – The encoder instance.
target (Any) – An optional pre-encoded schema.
kwargs – Additional named arguments.

Returns

The encoded schema of the instance.

Return type

Any

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

classmethod from_avro(schema, data)

Build a new instance from the avro schema and data.

Parameters

schema (Any) – The avro schema of the model.
data (Any) – The avro data of the model.

Returns

A new DynamicModel instance.

Return type

T

classmethod from_avro_bytes(raw, batch_mode=False, **kwargs)

Build a single instance or a sequence of instances from bytes.

Parameters

raw (bytes) – A bytes representation of the model.
batch_mode (bool) – If True the data is processed as a list of models, otherwise the data is processed as a single model.
kwargs – Additional named arguments.

Returns

A single instance or a sequence of instances.

Return type

Union[T, list[~T]]

classmethod from_avro_str(raw, **kwargs)

Build a single instance or a sequence of instances from bytes

Parameters: raw (str) – A str representation of the model.
Returns: A single instance or a sequence of instances.
Return type: Union[T, list[~T]]

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

Build a DeclarativeModel from a ModelType.

Parameters

model_type (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.

Returns

A new DeclarativeModel instance.

Return type

T

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

Build a Model from a TypeDict and data.

Parameters

typed_dict (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.

Returns

A new DataTransferObject instance.

Return type

T

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

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

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

model_type: alias of Event

classmethod to_avro_bytes(models)

Create a bytes representation of the given object instances.

Parameters: models (list[~T]) – A sequence of minos models.
Returns: A bytes object.
Return type: bytes

classmethod to_avro_str(models)

Build the avro string representation of the given object instances.

Parameters: models (list[~T]) – A sequence of minos models.
Returns: A bytes object.
Return type: str

type_hints = {'action': <enum 'Action'>, 'created_at': <class 'datetime.datetime'>, 'fields_diff': <class 'minos.aggregate.events.fields.FieldDiffContainer'>, 'name': <class 'str'>, 'uuid': <class 'uuid.UUID'>, 'version': <class 'int'>}

type_hints_parameters = ()

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