eventbus.util package

Submodules

eventbus.util.hashing module

eventbus.util.hashing.hash_object(json_encoder: eventbus.util.transcoding.ObjectJSONEncoder, obj: dict) → str

Calculates SHA-256 hash of JSON encoded ‘obj’.

eventbus.util.topic module

eventbus.util.topic.get_entity(event: TEvent) → TEntity

Get Originator Entity by event.

To support get_entity, event should provice originator_topic and __entity_kwargs__ properties Only Created event supports this protocol out of the box

Parameters:event – Domain event Returns:Domain entity

eventbus.util.topic.get_topic(domain_class: type) → str

Returns a string describing a class.

Parameters:domain_class – A class. Returns:A string describing the class.

eventbus.util.topic.resolve_attr(obj: Any, path: str) → Any

A recursive version of getattr for navigating dotted paths.

Parameters:

• obj – An object for which we want to retrieve a nested attribute.
• path – A dot separated string containing zero or more attribute names.

Raises:

AttributeError – If there is no such attribute.

Returns:

The attribute referred to by the path.

eventbus.util.topic.resolve_topic(topic: str) → Any

Resolves topic to the object it references.

Parameters:topic – A string describing a code object (e.g. an object class). Raises:TopicResolutionError – If there is no such class. Returns:Code object that the topic references.

eventbus.util.transcoding module

class eventbus.util.transcoding.ObjectJSONDecoder(object_hook=None, **kwargs)

Bases: json.decoder.JSONDecoder

class eventbus.util.transcoding.ObjectJSONEncoder(sort_keys=False)

Bases: json.encoder.JSONEncoder

encode(o) → bytes

Return a JSON string representation of a Python data structure.

>>> from json.encoder import JSONEncoder
>>> JSONEncoder().encode({"foo": ["bar", "baz"]})
'{"foo": ["bar", "baz"]}'

encode_container(o)

encode_deque(o)

encode_dict(o)

encode_dict_state(o)

encode_frozenset(o)

encode_instance(o)

encode_iterable(o)

encode_list(o)

encode_object(o)

encode_set(o)

encode_tuple(o)

eventbus.util.transcoding.b64str_from_bytes(value: bytes) → str

eventbus.util.transcoding.bytes_from_b64str(value)

eventbus.util.transcoding.decode_bytes(d)

eventbus.util.transcoding.decode_date(d)

eventbus.util.transcoding.decode_datetime(d)

eventbus.util.transcoding.decode_decimal(d)

eventbus.util.transcoding.decode_deque(d)

eventbus.util.transcoding.decode_dict(d)

eventbus.util.transcoding.decode_enum(d)

eventbus.util.transcoding.decode_frozenset(d)

eventbus.util.transcoding.decode_object(d)

eventbus.util.transcoding.decode_set(d)

eventbus.util.transcoding.decode_time(d)

eventbus.util.transcoding.decode_tuple(d)

eventbus.util.transcoding.decode_type(d)

eventbus.util.transcoding.decode_uuid(d)

eventbus.util.transcoding.decoder(d)

eventbus.util.transcoding.decoderpolicy(arg=None)

Decorator for decoder policy.

Allows default behaviour to be built up from methods registered for different named keys, rather than chain of “in dict” queries in a long if-else block.

eventbus.util.transcoding.encode_bytes(o)

eventbus.util.transcoding.encode_date(obj)

eventbus.util.transcoding.encode_datetime(obj)

eventbus.util.transcoding.encode_decimal(obj)

eventbus.util.transcoding.encode_enum(obj)

eventbus.util.transcoding.encode_function(o)

eventbus.util.transcoding.encode_method(o)

eventbus.util.transcoding.encode_time(obj)

eventbus.util.transcoding.encode_type(o)

eventbus.util.transcoding.encode_uuid(obj)

eventbus.util.transcoding.encoder(o)

eventbus.util.transcoding.encoderpolicy(arg=None)

Decorator for encoder policy.

Allows default behaviour to be built up from methods registered for different types of things, rather than chain of isinstance() calls in a long if-else block.

Module contents