Events Class

Events interface

Direct access to the events endpoint.

The user is not expected to use this class directly. It is an attribute of the Archivist class.

For example instantiate an Archivist instance and execute the methods of the class:

with open(".auth_token", mode="r", encoding="utf-8") as tokenfile:
    authtoken = tokenfile.read().strip()

# Initialize connection to Archivist
arch = Archivist(
    "https://app.datatrails.ai",
    authtoken,
)
asset = arch.assets.create(...)
event = arch.events.create(asset['identity'], ...)
class archivist.events.Event[source]

Event object has dictionary attributes and properties.

property when: str | None

Timestamp of event

property who: str | None

Principal identity.

class archivist.events._EventsPublic(archivist_instance: Archivist)[source]

EventsPublic

Access to events entities using the CRUD interface. This class is usually accessed as an attribute of the Archivist class.

Parameters:

archivist (Archivist) -- Archivist instance

_identity(identity: str) → str[source]

Return fully qualified identity If public then expect a full url as argument

count(*, asset_id: str | None = None, props: dict[str, Any] | None = None, attrs: dict[str, Any] | None = None, asset_attrs: dict[str, Any] | None = None) → int[source]

Count events.

Counts number of events that match criteria.

Parameters:

  • asset_id (str) -- optional asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx

  • props (dict) -- optional properties e.g. {"confirmation_status": "CONFIRMED" }

  • attrs (dict) -- optional attributes e.g. {"arc_display_type": "open" }

  • asset_attrs (dict) -- optional asset_attributes e.g. {"arc_display_type": "door" }

Returns:

integer count of assets.

list(*, asset_id: str | None = None, page_size: int | None = None, props: dict[str, Any] | None = None, attrs: dict[str, Any] | None = None, asset_attrs: dict[str, Any] | None = None)[source]

List events.

Lists events that match criteria.

Parameters:

  • asset_id (str) -- optional asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx

  • props (dict) -- e.g. {"tracked": "TRACKED" }

  • attrs (dict) -- e.g. {"arc_display_type": "open" }

  • asset_attrs (dict) -- optional asset_attributes e.g. {"arc_display_type": "door" }

  • page_size (int) -- optional page size. (Rarely used).

Returns:

iterable that returns Event instances

read(identity: str) → Event[source]

Read event

Reads event.

Parameters:

identity (str) -- events identity e.g. assets/xxxxxxx.../events/yyyyyyy...

Returns:

Event instance

read_by_signature(*, asset_id: str | None = None, props: dict[str, Any] | None = None, attrs: dict[str, Any] | None = None, asset_attrs: dict[str, Any] | None = None) → Event[source]

Read event by signature.

Reads event that meets criteria. Only one event is expected.

Parameters:

  • asset_id (str) -- optional asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx

  • props (dict) -- e.g. {"tracked": "TRACKED" }

  • attrs (dict) -- e.g. {"arc_display_type": "open" }

  • asset_attrs (dict) -- optional asset_attributes e.g. {"arc_display_type": "door" }

Returns:

Event instance

class archivist.events._EventsRestricted(archivist_instance: Archivist)[source]

EventsRestricted

Access to events entities using the CRUD interface. This class is usually accessed as an attribute of the Archivist class.

Parameters:

archivist (Archivist) -- Archivist instance

create(asset_id: str, props: dict[str, Any], attrs: dict[str, Any], *, asset_attrs: dict[str, Any] | None = None, confirm: bool = False) → Event[source]

Create event

Creates event for given asset.

Parameters:

  • asset_id (str) -- asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx

  • props (dict) -- properties for this event.

  • attrs (dict) -- attributes of created event.

  • asset_attrs (dict) -- attributes of referenced asset.

  • confirm (bool) -- if True wait for event to be confirmed.

Returns:

Event instance

create_from_data(asset_id: str, data: dict[str, Any], *, confirm: bool = False) → Event[source]

Create event

Creates event for given asset from data. Suitable for reading data from json.load or yaml.load from a file

Parameters:

  • asset_id (str) -- asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx

  • data (dict) -- request body of event.

  • confirm (bool) -- if True wait for event to be confirmed.

Returns:

Event instance

publicurl(identity: str) → str[source]

Read event public url

Reads event public url.

Parameters:

identity (str) -- events identity e.g. assets/xxxxxxx.../events/yyyyyyy...

Returns:

str:public url as a string

wait_for_confirmation(identity: str) → Event[source]

Wait for event to be confirmed.

Waits for event to be confirmed.

Parameters:

identity (str) -- identity of event

Returns:

True if event is confirmed.

wait_for_confirmed(*, asset_id: str | None = None, props: dict[str, Any] | None = None, attrs: dict[str, Any] | None = None, asset_attrs: dict[str, Any] | None = None) → bool[source]

Wait for events to be confirmed.

Waits for all events that match criteria to be confirmed.

Parameters:

  • asset_id (str) -- optional asset identity e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx

  • props (dict) -- e.g. {"tracked": "TRACKED" }

  • attrs (dict) -- e.g. {"arc_display_type": "open" }

  • asset_attrs (dict) -- optional asset_attributes e.g. {"arc_display_type": "door" }

Returns:

True if all events are confirmed.