hopeit.dataobjects package

Data Objects type abstractions. A Dataobject is basically any object wrapped in a Dataclass and that is able to be stored, retrieved, serialized and deserialized by the platform toolkit.

Annotate dataclasses with @dataobject annotation to make it support:
  • JSON schema generation, validation and serialization/deserialization
  • Compression and Serialization using different mechanisms

Example:

from hopeit.dataobjects import dataclass, dataobject

@dataobject @dataclass class MyObject:

name: str number: int
class hopeit.dataobjects.StreamEventParams(event_id_expr: Optional[str], event_ts_expr: Optional[str])

Bases: object

Helper class used to access attributes in @dataobject decorated objects, based on dot notation expressions

static extract_attr(obj, expr)
hopeit.dataobjects.dataobject(decorated_class=None, *, event_id: Optional[str] = None, event_ts: Optional[str] = None, unsafe: bool = False, validate: bool = True, schema: bool = True)

Decorator for dataclasses intended to be used in API and/or streams. This decorated mainly implements JsonSchemaMixIn adding dataclass functionality to:

  • Generate Json Schema for Open API definition
  • Parse and convert from and to json
  • Validate Json against Json schema
  • Detect incompatibilities between API specification and payload structures
  • Detect undocumented API changes early (i.e. fields added or changed)

In general, all dataclasses that are to be exchanged using API endpoints (payload and responses), or write and read from streams, need to implement @dataobject and decorator in addition to Python @dataclass decorator.

In order to publish instances to streams, an event id and event timestamp can be extracted. StreamManager does that automatically for classes defining event_id() and event_ts() methods.

This decorator, adds these two methods to a dataclass: event_id(): str, extract the id of the object from a given dot notation expression event_ts(): Optional[datetime], extract timestamp from a given dot notation expression

Parameters:
  • decorated_class – decorated class
  • event_id – optional str, dot notation expression to navigate to id field (i.e. ‘id’ or ‘event.id’)
  • event_ts – optional str, dot notation expression to navigate to a datetime field (i.e. ‘last_update.ts’)
  • unsafe – bool, default False. When False, every time a new step is invoked a copy of dataobject will be sent, preventing object to be mutated unintentionally. Specifying unsafe=True prevents making copies every time a step is invoked, improving performance. Use with caution: with unsafe=True, you can accidentally mutate objects after they are yield, specially when returning generators or Spawn[…].
  • validate – bool, default True: indicates whether to validate using JsonSchema automatically generated by @dataobject annotation when reading from and converting to json using Json.from_json and Json.to_json. Notice that if you call from_json or to_json directly from annotated dataobject you need to specify validate=True/False parameter value.
  • schema – bool, default True: indicates to attempt json_schema generation in API module

In case event_id is not provided, an uuid will be generated on each call to event_id() In case event_ts is not provided, None will be returned on each call to event_ts()

Example:

@dataobject
@dataclass
class StatusChange:
    ts: datetime
    status: str

@dataobject(event_id='id', event_ts='last_status.ts', unsafe=True, validate=False)
@dataclass
class EventData:
    id: str
    last_status: StatusChange
hopeit.dataobjects.copy_payload(original: Union[str, int, float, bool, dict, set, list, DataObject, None]) → Union[str, int, float, bool, dict, set, list, DataObject, None]

Creates a copy of the original DataObject in case it is mutable. Returns original object in case it is a frozen dataclass

Submodules

Json tools to serialized and deserialze data objects

class hopeit.dataobjects.jsonify.Json

Bases: typing.Generic

Json convenience ser/deser functions for @dataobject decorated object (@see DataObject)

static from_json(json_str: Union[str, bytes], datatype: Type[EventPayloadType], key: str = 'value') → EventPayloadType

Converts json_str to desired datatype

Parameters:
  • json_str – str containing valid json, or string representation for atomic values
  • datatype – supported types defined in EventPayload
  • key – key to extract atomic types from
Returns:

instance of datatype
static parse_form_field(field_data: Union[str, dict], datatype: Type[EventPayloadType], key: str = 'value') → EventPayloadType

Helper to parse dataobjects from form-fields where encoding type is not correctly set to json

static to_json(payload: EventPayloadType, key: Optional[str] = 'value') → str

Converts event payload to json string

Parameters:
  • payload – EventPayload, instance of supported object type
  • key – key name used in generated json when serializing atomic values
Returns:

str containing json representation of data. In case of simple datatypes, a json str of key:value form will be generated using key parameter if it’s not None.