minos.saga.context module

class minos.saga.context.SagaContext(**kwargs)[source]

Bases: minos.common.model.dynamic.bucket.BucketModel, collections.abc.MutableMapping

Saga Context class

The purpose of this class is to keep an execution state.

__init__(**kwargs)[source]

Class constructor.

Parameters

fields – Dictionary that contains the Field instances of the model indexed by name.

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': 'SagaContext', 'namespace': 'minos.saga.context.913089f6-2ed6-4426-ba6b-982b1e13dae0', 'type': 'record', 'fields': []}]
property avro_str: str

Generate bytes representation of the current instance.

Return type

str

Returns

A bytes object.

classname = 'minos.saga.context.SagaContext'
clear() → None.  Remove all items from D.
classmethod empty()

Build an empty BucketModel instance.

Return type

~T

Returns

A BucketModel instance.

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 DynamicModel 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 DynamicModel 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.
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 minos.common.model.types.model_types.SagaContext

pop(k[, d]) → v, remove specified key and return the corresponding value.

If key is not found, d is returned if given, otherwise KeyError is raised.

popitem() → (k, v), remove and return some (key, value) pair

as a 2-tuple; but raise KeyError if D is empty.

setdefault(k[, d]) → D.get(k,d), also set D[k]=d if k not in D
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.

type_hints = {}
type_hints_parameters = ()
update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

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