Skip to content
Roboto
Esc
↑↓navigate↵open⌘Jpreview
On this page

roboto.domain.platform_events

Platform events: what happened to an entity on the platform, as a record.

Distinct from roboto.domain.events, whose Event is a time-anchored annotation on robotics data. A platform event is emitted by the platform when a file is uploaded, a dataset is created, an invocation finishes, and so on. Triggers subscribe to them; outgoing integrations receive them as CloudEvents. The catalog describes each type: its payload model, the namespace roots it exposes, and the once_per values it supports.

Submodules

Package Contents

CLOUDEVENTS_SPECVERSION

roboto.domain.platform_events.CLOUDEVENTS_SPECVERSION = '1.0'#View Source

The CloudEvents spec version PlatformEvent.to_cloudevent() produces.

CLOUDEVENTS_TYPE_PREFIX

roboto.domain.platform_events.CLOUDEVENTS_TYPE_PREFIX = 'ai.roboto.'#View Source

Reverse-DNS prefix CloudEvents recommends on type. It appears only in what PlatformEvent.to_cloudevent() returns; subscriptions, conditions, and templates name an event by its bare PlatformEventType value (file.uploaded).

DATASET_ROOT

roboto.domain.platform_events.DATASET_ROOT = 'dataset'#View Source

Namespace root naming a dataset. An event exposing it is about a dataset or about something within one, so a consumer may reach that dataset’s files from the event.

DEFAULT_PLATFORM_EVENT_CATALOG

roboto.domain.platform_events.DEFAULT_PLATFORM_EVENT_CATALOG#View Source

The catalog of every PlatformEventType, used wherever a caller does not supply its own (envelope payload binding, trigger validation, evaluation).

DatasetCreatedPayload

class roboto.domain.platform_events.DatasetCreatedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.DatasetCreated.

Parameters

data Any

Attributes

DatasetCreatedPayload.dataset_id

dataset_id str #

The created dataset.

DatasetCreatedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

DatasetMetadataUpdatedPayload

class roboto.domain.platform_events.DatasetMetadataUpdatedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.DatasetMetadataUpdated.

Parameters

data Any

Attributes

DatasetMetadataUpdatedPayload.changeset

The applied metadata/tag delta, served to conditions as the changed and tag roots.

DatasetMetadataUpdatedPayload.dataset_id

dataset_id str #

The dataset whose metadata changed.

DatasetMetadataUpdatedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

DatasetTagAddedPayload

class roboto.domain.platform_events.DatasetTagAddedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.DatasetTagAdded.

Parameters

data Any

Attributes

DatasetTagAddedPayload.dataset_id

dataset_id str #

The tagged dataset.

DatasetTagAddedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

DatasetTagAddedPayload.tags_added

tags_added list[str] #

Tags added by the mutation, served to conditions as the tag root.

ENVELOPE_ROOT

roboto.domain.platform_events.ENVELOPE_ROOT = 'envelope'#View Source

Namespace root served from the platform event’s own envelope: its id, source, subject, type, time, and org, plus the payload as published under data. Reserved: no event type may expose it as an entity root.

EventCreatedPayload

class roboto.domain.platform_events.EventCreatedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.EventCreated.

Parameters

data Any

Attributes

EventCreatedPayload.event_id

event_id str #

The created event, an EventRecord.

EventCreatedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

FileIngestedPayload

class roboto.domain.platform_events.FileIngestedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.FileIngested.

Parameters

data Any

Attributes

FileIngestedPayload.dataset_id

dataset_id str #

Dataset containing the ingested file.

FileIngestedPayload.file_id

file_id str #

The ingested file.

FileIngestedPayload.file_version

file_version int | None = None #

The file’s version when the event was published; a newer version may exist by the time a trigger evaluates. Versions number revisions of the file record, which a metadata or tag edit advances just as an overwrite of the file’s contents does. None on events published before the payload carried this field.

FileIngestedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

FileIngestedPayload.transaction_id

transaction_id str | None = None #

Upload transaction the file arrived in, when known.

FileMetadataUpdatedPayload

class roboto.domain.platform_events.FileMetadataUpdatedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.FileMetadataUpdated.

Parameters

data Any

Attributes

FileMetadataUpdatedPayload.changeset

The applied metadata/tag delta, served to conditions as the changed and tag roots.

FileMetadataUpdatedPayload.dataset_id

dataset_id str #

Dataset containing the updated file.

FileMetadataUpdatedPayload.file_id

file_id str #

The file whose metadata changed.

FileMetadataUpdatedPayload.file_version

file_version int | None = None #

The file’s version when the event was published; a newer version may exist by the time a trigger evaluates. Versions number revisions of the file record, which a metadata or tag edit advances just as an overwrite of the file’s contents does. None on events published before the payload carried this field.

FileMetadataUpdatedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

FileUploadedPayload

class roboto.domain.platform_events.FileUploadedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.FileUploaded.

Parameters

data Any

Attributes

FileUploadedPayload.dataset_id

dataset_id str #

Dataset the file was uploaded to.

FileUploadedPayload.file_id

file_id str #

The uploaded file.

FileUploadedPayload.file_version

file_version int | None = None #

The file’s version when the event was published; a newer version may exist by the time a trigger evaluates. Versions number revisions of the file record, which a metadata or tag edit advances just as an overwrite of the file’s contents does. None on events published before the payload carried this field.

FileUploadedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

FileUploadedPayload.transaction_id

transaction_id str | None = None #

Upload transaction the file arrived in, when the upload used one.

InvocationCompletedPayload

class roboto.domain.platform_events.InvocationCompletedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.InvocationCompleted.

Parameters

data Any

Attributes

InvocationCompletedPayload.action_name

action_name str #

Name of the invoked action. Unique within action_owner_id’s org.

InvocationCompletedPayload.action_owner_id

action_owner_id str #

Org that owns the invoked action.

InvocationCompletedPayload.invocation_id

invocation_id str #

The completed invocation.

InvocationCompletedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

InvocationCompletedPayload.status

Terminal status the invocation reached.

InvocationFailedPayload

class roboto.domain.platform_events.InvocationFailedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.InvocationFailed.

Parameters

data Any

Attributes

InvocationFailedPayload.action_name

action_name str #

Name of the invoked action. Unique within action_owner_id’s org.

InvocationFailedPayload.action_owner_id

action_owner_id str #

Org that owns the invoked action.

InvocationFailedPayload.invocation_id

invocation_id str #

The failed invocation.

InvocationFailedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

InvocationFailedPayload.status

Terminal status the invocation reached (Failed or Deadly).

OncePer

class roboto.domain.platform_events.OncePer#View Source

Bases: roboto.compat.StrEnum

A grain an event offers for deduplication: the occurrence itself, or an entity the event names.

This is an event’s vocabulary, not a trigger’s. Each PlatformEventType declares the grains it supports, and how an occurrence projects onto each, in its PlatformEventDescriptor; that projection is what turns one occurrence into one idempotency token. A trigger only chooses among the grains its events offer, and its choice must be legal for every event type it subscribes to.

Attributes

OncePer.Dataset

Dataset = 'dataset' #

Fire once per dataset.

OncePer.Event

Event = 'event' #

Fire once per event, the annotation marking a span of time on your data.

OncePer.File

File = 'file' #

Fire once per file.

OncePer.Invocation

Invocation = 'invocation' #

Fire once per action invocation.

OncePer.Occurrence

Occurrence = 'occurrence' #

Fire on every occurrence; nothing collapses. Supported by every event type.

OncePerProjection

roboto.domain.platform_events.OncePerProjection#View Source

Projects a PlatformEvent onto the token string that identifies what the trigger fires once per (e.g. the dataset id for once_per=dataset).

A projection reads the event payload and nothing else: one event in, one string out, with no lookup against platform state. A once_per value can only name what the event itself already identifies, so there is no “once per file in the dataset this event is about”. once_per only collapses repeats; it never fans one event out into several runs, and a trigger dispatches each of its targets at most once per event.

PlatformEvent

class roboto.domain.platform_events.PlatformEvent(/, **data)#View Source

Bases: pydantic.BaseModel

One occurrence of something that happened on the platform, as a trigger receives it.

id, source, type, time and subject are the CloudEvents 1.0 context attributes; to_cloudevent() renders the occurrence in that spec’s structured-JSON form. The envelope stays thin: data carries entity ids and facts fixed at time, such as a metadata delta or a terminal status, and never mutable entity state — conditions and target templates read an entity’s current state at evaluation time through an EventNamespace.

Parameters

data Any

Attributes

PlatformEvent.data

The per-type payload. Its model must be the one registered for type in the default catalog; a mismatch is rejected at validation time.

PlatformEvent.id

id str #

Producer-vended event id, unique within source; deterministic where possible (e.g. evt:{invocation_id}:completed).

PlatformEvent.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

PlatformEvent.org_id

org_id str #

Organization in which the event occurred and whose triggers see it.

PlatformEvent.schema_version

schema_version int = 1 #

Version of the envelope + payload schema. Bumped only for breaking changes.

PlatformEvent.source

source str #

The org and the deployment the event happened in, as built by platform_event_source().

Properties

PlatformEvent.subject

subject str #

The entity the event is about, as a roboto:// URI (roboto://file/fl_1a2b).

Computed from data by the event type’s catalog descriptor rather than stored, so it cannot disagree with the payload, and serialized like a declared field.

Return type: str

PlatformEvent.subject_uri

subject as a parsed RobotoUri.

Attributes

PlatformEvent.time

time datetime.datetime #

When the event occurred.

PlatformEvent.to_cloudevent()

to_cloudevent()#View Source

Return this event as CloudEvents 1.0 structured JSON (application/cloudevents+json).

Nothing serializes it yet: the command bus carries the model itself. It exists so an outbound webhook body needs no redesign.

Context attributes sit at the top level under their spec names, with type carrying CLOUDEVENTS_TYPE_PREFIX and datacontenttype fixed at application/json. org_id and schema_version arrive as the extension attributes orgid and schemaversion, because extension names must be lowercase alphanumerics. The returned dict is JSON-serializable as-is.

Return type

dict[str, Any]

Attributes

PlatformEvent.type

Which kind of event this is. Dictates the concrete model of data.

PlatformEventCatalog

class roboto.domain.platform_events.PlatformEventCatalog(descriptors)#View Source

Descriptors for platform event types, looked up by type.

Everything that varies by event type — payload model, exposed roots, default root, dedup projections — hangs off the PlatformEventDescriptor objects registered here. DEFAULT_PLATFORM_EVENT_CATALOG registers every member of PlatformEventType; a caller may build a catalog over a subset.

Parameters

descriptors collections.abc.Iterable[PlatformEventDescriptor]

PlatformEventCatalog.descriptor()

descriptor(event_type)#View Source

Return the descriptor for event_type.

Raises

ValueError

No descriptor is registered for event_type.

PlatformEventCatalog.exposed_roots()

exposed_roots(event_type)#View Source

Return the namespace roots event_type exposes.

Raises

ValueError

No descriptor is registered for event_type.

Return type

frozenset[str]

PlatformEventCatalog.namespace_roots()

namespace_roots()#View Source

Return the union of every registered event type’s exposed roots.

Return type

frozenset[str]

PlatformEventCatalog.subscribable_types()

subscribable_types()#View Source

Return the event types a trigger may subscribe to.

PlatformEventDescriptor

class roboto.domain.platform_events.PlatformEventDescriptor#View Source

Everything the trigger system knows about one platform event type.

Immutable. Carries the payload model the envelope validates against, the namespace roots the event exposes to conditions and target templates, the default root unqualified condition fields bind to, the entity the event is about, and the OncePer values the event supports (as projections from an event to its dedup token value).

Every once_per value is declared as an OncePerProjection, a pure function of the event payload, so a descriptor can only offer what the payload itself names. No descriptor can resolve one event into several subjects, which is what holds dispatch at one per target per event.

Attributes

PlatformEventDescriptor.default_root

default_root str #

Root that unqualified condition fields bind to. Always one of exposed_roots.

PlatformEventDescriptor.event_type

The event type this descriptor describes.

PlatformEventDescriptor.exposed_roots

exposed_roots frozenset[str] #

Namespace roots (dataset, file, changed, …) this event exposes.

PlatformEventDescriptor.idempotency_token()

idempotency_token(event, once_per)#View Source

Return the dedup token for event at once_per.

Two events that project onto the same token dispatch the same target of the same trigger at most once.

Parameters

The event to project. Must be of this descriptor’s type.

What the trigger fires once per.

Raises

ValueError

event is of a different type, or once_per is not supported by this event type.

Return type

str

Attributes

PlatformEventDescriptor.once_per_projections

once_per_projections collections.abc.Mapping[roboto.domain.platform_events.once_per.OncePer, OncePerProjection] #

Supported once_per values, each mapped to the projection that yields its token. Key presence defines legality; every event supports OncePer.Occurrence.

PlatformEventDescriptor.payload_model

payload_model type[pydantic.BaseModel] #

Model of PlatformEvent.data for this event type.

PlatformEventDescriptor.subject()

subject(event)#View Source

Return the roboto:// URI of the entity event is about.

Raises

ValueError

event is not of this descriptor’s event type.

Attributes

PlatformEventDescriptor.subject_type

The kind of entity this event is about — its CloudEvents subject. The payload carries that entity’s id under {subject_type}_id, and may say more about it, such as the version a file was at. The rest is context around the entity: the upload transaction a file arrived in, or the applied changeset.

PlatformEventDescriptor.subscribable

subscribable bool = True #

Whether a trigger may name this type in an EventSubscription. False for occurrences delivered to a single trigger rather than broadcast to subscribers — the schedule tick — which still carry a descriptor so their tokens and namespaces are built the same way.

Properties

PlatformEventDescriptor.supported_once_per

The OncePer values this event type supports.

PlatformEventDescriptor.supports_once_per()

supports_once_per(once_per)#View Source

Return whether once_per is legal for this event type.

Return type

bool

PlatformEventPayload

roboto.domain.platform_events.PlatformEventPayload#View Source

Union of every per-type payload model. Which member is legal for a given PlatformEvent is dictated by its type.

PlatformEventType

class roboto.domain.platform_events.PlatformEventType#View Source

Bases: roboto.compat.StrEnum

A thing that happens on the platform that triggers can subscribe to.

Every member has a PlatformEventDescriptor in the default PlatformEventCatalog describing its payload model, the namespace roots it exposes to conditions and templates, and the OncePer values it supports.

Attributes

PlatformEventType.DatasetCreated

DatasetCreated = 'dataset.created' #

A dataset was created.

PlatformEventType.DatasetMetadataUpdated

DatasetMetadataUpdated = 'dataset.metadata_updated' #

A dataset’s metadata or tags changed.

PlatformEventType.DatasetTagAdded

DatasetTagAdded = 'dataset.tag_added' #

One or more tags were added to a dataset.

PlatformEventType.EventCreated

EventCreated = 'event.created' #

An event, the annotation marking a span of time on your data, was created.

PlatformEventType.FileIngested

FileIngested = 'file.ingested' #

A file finished ingestion (post-processing) and its topics are available.

PlatformEventType.FileMetadataUpdated

FileMetadataUpdated = 'file.metadata_updated' #

A file’s metadata or tags changed.

PlatformEventType.FileUploaded

FileUploaded = 'file.uploaded' #

A file finished uploading to a dataset.

PlatformEventType.InvocationCompleted

InvocationCompleted = 'invocation.completed' #

An action invocation reached a successful terminal status.

PlatformEventType.InvocationFailed

InvocationFailed = 'invocation.failed' #

An action invocation reached a failed terminal status (Failed or Deadly).

PlatformEventType.ScheduleFired

ScheduleFired = 'schedule.fired' #

A trigger’s own schedule reached one of its minutes. Not subscribable: a trigger fires on a schedule by declaring a Schedule source, and the scheduler delivers this occurrence to that trigger alone.

PlatformEventType.SessionIngested

SessionIngested = 'session.ingested' #

Every ingestable file in a complete session is ingested. Fires again, as a new occurrence, each time the session’s ingestable files change and are all ingested again.

PlatformEventType.UploadCompleted

UploadCompleted = 'dataset.upload_completed' #

An upload transaction to a dataset completed (all of its files uploaded).

Properties

PlatformEventType.cloudevents_type

cloudevents_type str #

This type as a CloudEvents type attribute: ai.roboto.file.uploaded.

Return type: str

PlatformEventType.from_cloudevents_type()

classmethod from_cloudevents_type(value)#View Source

Return the event type a CloudEvents type attribute names; the inverse of cloudevents_type.

Parameters

value str

A prefixed type, such as ai.roboto.file.uploaded.

Raises

ValueError

value lacks the prefix or names no platform event type.

RESERVED_ROOTS

roboto.domain.platform_events.RESERVED_ROOTS#View Source

Roots the evaluation namespace serves itself. A descriptor claiming one would be silently shadowed by the envelope or the trigger, so construction rejects it.

SAMPLE_ACTION_DIGEST

roboto.domain.platform_events.SAMPLE_ACTION_DIGEST = 'sha256:9f1c2e7b4a0d8c6f3e5b1a7d2c4f8e6a0b3d5c7e9f1a2b4c6d8e0f2a4b6c8d0e'#View Source

SAMPLE_ACTION_NAME

roboto.domain.platform_events.SAMPLE_ACTION_NAME = 'ros-ingest'#View Source

SAMPLE_API_DOMAIN

roboto.domain.platform_events.SAMPLE_API_DOMAIN = 'api.roboto.ai'#View Source

SAMPLE_CHANGESET

roboto.domain.platform_events.SAMPLE_CHANGESET#View Source

SAMPLE_DATASET_ID

roboto.domain.platform_events.SAMPLE_DATASET_ID = 'ds_7h2k9m4qxp3w'#View Source

SAMPLE_EVENT_ID

roboto.domain.platform_events.SAMPLE_EVENT_ID = 'ev_7g3n5kq2wxrd'#View Source

SAMPLE_FILE_ID

roboto.domain.platform_events.SAMPLE_FILE_ID = 'fl_q8v2n6ty4mcs'#View Source

SAMPLE_FILE_VERSION

roboto.domain.platform_events.SAMPLE_FILE_VERSION = 2#View Source

SAMPLE_INVOCATION_ID

roboto.domain.platform_events.SAMPLE_INVOCATION_ID = 'iv_2x9pd7wk5rhf'#View Source

SAMPLE_ORG_ID

roboto.domain.platform_events.SAMPLE_ORG_ID = 'og_k3m8w2rq7nxd'#View Source

SAMPLE_SESSION_ID

roboto.domain.platform_events.SAMPLE_SESSION_ID = 'se_m4t7c1zq9bvn'#View Source

SAMPLE_TAGS_ADDED

roboto.domain.platform_events.SAMPLE_TAGS_ADDED = ['validated', 'nightly']#View Source

SAMPLE_TIME

roboto.domain.platform_events.SAMPLE_TIME#View Source

The instant every sample event and record is dated from, so samples are stable across calls.

SAMPLE_TRANSACTION_ID

roboto.domain.platform_events.SAMPLE_TRANSACTION_ID = 'tx_5jw8r3ne2kpq'#View Source

SAMPLE_TRIGGER_ID

roboto.domain.platform_events.SAMPLE_TRIGGER_ID = 'tr_3nq7wk2mx9pd'#View Source

SAMPLE_USER

roboto.domain.platform_events.SAMPLE_USER = 'maria.chen@acme-robotics.com'#View Source

ScheduleFiredPayload

class roboto.domain.platform_events.ScheduleFiredPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.ScheduleFired.

Parameters

data Any

Attributes

ScheduleFiredPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ScheduleFiredPayload.scheduled_for

scheduled_for datetime.datetime #

The scheduled minute (UTC) this occurrence stands for.

ScheduleFiredPayload.trigger_id

trigger_id str #

The trigger whose schedule fired.

SessionIngestedPayload

class roboto.domain.platform_events.SessionIngestedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.SessionIngested.

Parameters

data Any

Attributes

SessionIngestedPayload.ingestion_number

ingestion_number int #

Which announcement this is for the session, starting at 1.

SessionIngestedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

SessionIngestedPayload.session_id

session_id str #

The session whose ingestable files are all ingested.

TRIGGER_ROOT

roboto.domain.platform_events.TRIGGER_ROOT = 'trigger'#View Source

Namespace root describing the trigger being evaluated. Reserved for the consumer; no event type may expose it as an entity root.

UploadCompletedPayload

class roboto.domain.platform_events.UploadCompletedPayload(/, **data)#View Source

Bases: pydantic.BaseModel

Payload for PlatformEventType.UploadCompleted.

Parameters

data Any

Attributes

UploadCompletedPayload.dataset_id

dataset_id str #

Dataset the upload targeted.

UploadCompletedPayload.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

UploadCompletedPayload.transaction_id

transaction_id str #

The completed upload transaction.

event_catalog_manifest()

roboto.domain.platform_events.event_catalog_manifest(catalog=DEFAULT_PLATFORM_EVENT_CATALOG)#View Source

The catalog as the web UI’s TypeScript mirror reads it: per event type, its roots, default root, supported grains, subject type, and whether a trigger may subscribe.

event_catalog.json in this package is this function’s output, written by scripts/gen_trigger_manifests.py and drift-checked by a test on each side.

Parameters

Return type

dict[str, Any]

platform_event_source()

roboto.domain.platform_events.platform_event_source(api_domain, org_id)#View Source

Return the CloudEvents source for one org’s events on one deployment.

The value is the org’s own API resource, https://{api_domain}/v1/orgs/{org_id}. No two deployments produce the same source, so it pairs with PlatformEvent.id to identify a single occurrence.

Parameters

api_domain Optional[str]

Public API host of the emitting deployment, such as api.roboto.ai. Without one, the result is the relative reference /v1/orgs/{org_id}.

org_id str

Organization whose events carry this source.

Return type

str

sample_platform_event()

roboto.domain.platform_events.sample_platform_event(event_type)#View Source

A valid, fully populated event of event_type from the sample scenario.

Was this page helpful?