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#

roboto.domain.platform_events.CLOUDEVENTS_SPECVERSION = '1.0'#: The CloudEvents spec version PlatformEvent.to_cloudevent() produces.

roboto.domain.platform_events.CLOUDEVENTS_TYPE_PREFIX = 'ai.roboto.'#: 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).

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

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.DatasetCreated.

Parameters:: data (Any)

dataset_id: str#: The created dataset.

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

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.DatasetMetadataUpdated.

Parameters:: data (Any)

changeset: roboto.updates.MetadataChangeset#: The applied metadata/tag delta, served to conditions as the changed and tag roots.

dataset_id: str#: The dataset whose metadata changed.

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

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.DatasetTagAdded.

Parameters:: data (Any)

dataset_id: str#: The tagged dataset.

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

tags_added: list[str]#: Tags added by the mutation, served to conditions as the tag root.

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

its id, source, subject, type, time, and org. Reserved: no event type may expose it as an entity root.

Type:: Namespace root served from the platform event’s own envelope

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.EventCreated.

Parameters:: data (Any)

event_id: str#: The created event, an EventRecord.

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

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.FileIngested.

Parameters:: data (Any)

dataset_id: str#: Dataset containing the ingested file.

file_id: str#: The ingested file.

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

transaction_id: str | None = None#: Upload transaction the file arrived in, when known.

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.FileMetadataUpdated.

Parameters:: data (Any)

changeset: roboto.updates.MetadataChangeset#: The applied metadata/tag delta, served to conditions as the changed and tag roots.

dataset_id: str#: Dataset containing the updated file.

file_id: str#: The file whose metadata changed.

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

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.FileUploaded.

Parameters:: data (Any)

dataset_id: str#: Dataset the file was uploaded to.

file_id: str#: The uploaded file.

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

transaction_id: str | None = None#: Upload transaction the file arrived in, when the upload used one.

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.InvocationCompleted.

Parameters:: data (Any)

action_name: str#: Name of the invoked action. Unique within action_owner_id’s org.

action_owner_id: str#: Org that owns the invoked action.

invocation_id: str#: The completed invocation.

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

status: roboto.domain.actions.invocation_record.InvocationStatus#: Terminal status the invocation reached.

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.InvocationFailed.

Parameters:: data (Any)

action_name: str#: Name of the invoked action. Unique within action_owner_id’s org.

action_owner_id: str#: Org that owns the invoked action.

invocation_id: str#: The failed invocation.

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

status: roboto.domain.actions.invocation_record.InvocationStatus#: Terminal status the invocation reached (Failed or Deadly).

class roboto.domain.platform_events.OncePer#

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.

Dataset = 'dataset'#: Fire once per dataset.

Event = 'event'#: Fire once per event, the annotation marking a span of time on your data.

File = 'file'#: Fire once per file.

Invocation = 'invocation'#: Fire once per action invocation.

Occurrence = 'occurrence'#: Fire on every occurrence; nothing collapses. Supported by every event type.

Session = 'session'#: Fire once per session.

roboto.domain.platform_events.OncePerProjection#

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.

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

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)

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

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

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

org_id: str#: Organization in which the event occurred and whose triggers see it.

schema_version: int = 1#: Version of the envelope + payload schema. Bumped only for breaking changes.

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

property 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

property subject_uri: roboto.uri.RobotoUri#

subject as a parsed RobotoUri.

Return type:: roboto.uri.RobotoUri

time: datetime.datetime#: When the event occurred.

to_cloudevent()#

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]

type: PlatformEventType#: Which kind of event this is. Dictates the concrete model of data.

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

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])

descriptor(event_type)#

Return the descriptor for event_type.

Raises:: ValueError – No descriptor is registered for event_type.
Parameters:: event_type (roboto.domain.platform_events.events.PlatformEventType)
Return type:: PlatformEventDescriptor

exposed_roots(event_type)#

Return the namespace roots event_type exposes.

Raises:: ValueError – No descriptor is registered for event_type.
Parameters:: event_type (roboto.domain.platform_events.events.PlatformEventType)
Return type:: frozenset[str]

namespace_roots()#

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

Return type:: frozenset[str]

subscribable_types()#

Return the event types a trigger may subscribe to.

Return type:: frozenset[roboto.domain.platform_events.events.PlatformEventType]

class roboto.domain.platform_events.PlatformEventDescriptor#

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.

default_root: str#: Root that unqualified condition fields bind to. Always one of exposed_roots.

event_type: roboto.domain.platform_events.events.PlatformEventType#: The event type this descriptor describes.

exposed_roots: frozenset[str]#: Namespace roots (dataset, file, changed, …) this event exposes.

idempotency_token(event, once_per)#

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:

event (roboto.domain.platform_events.events.PlatformEvent) – The event to project. Must be of this descriptor’s type.
once_per (roboto.domain.platform_events.once_per.OncePer) – 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

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.

payload_model: type[pydantic.BaseModel]#: Model of PlatformEvent.data for this event type.

subject(event)#

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

Raises:: ValueError – event is not of this descriptor’s event type.
Parameters:: event (roboto.domain.platform_events.events.PlatformEvent)
Return type:: roboto.uri.RobotoUri

subject_type: roboto.uri.RobotoUriType#: The kind of entity this event is about — its CloudEvents subject. The payload carries that entity’s id under {subject_type}_id; everything else in it is context around the entity: the upload transaction a file arrived in, the files added to a session, the applied changeset.

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.

property supported_once_per: frozenset[roboto.domain.platform_events.once_per.OncePer]#

The OncePer values this event type supports.

Return type:: frozenset[roboto.domain.platform_events.once_per.OncePer]

supports_once_per(once_per)#

Return whether once_per is legal for this event type.

Parameters:: once_per (roboto.domain.platform_events.once_per.OncePer)
Return type:: bool

roboto.domain.platform_events.PlatformEventPayload#: Union of every per-type payload model. Which member is legal for a given PlatformEvent is dictated by its type.

class roboto.domain.platform_events.PlatformEventType#

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.

DatasetCreated = 'dataset.created'#: A dataset was created.

DatasetMetadataUpdated = 'dataset.metadata_updated'#: A dataset’s metadata or tags changed.

DatasetTagAdded = 'dataset.tag_added'#: One or more tags were added to a dataset.

EventCreated = 'event.created'#: An event, the annotation marking a span of time on your data, was created.

FileIngested = 'file.ingested'#: A file finished ingestion (post-processing) and its topics are available.

FileMetadataUpdated = 'file.metadata_updated'#: A file’s metadata or tags changed.

FileUploaded = 'file.uploaded'#: A file finished uploading to a dataset.

InvocationCompleted = 'invocation.completed'#: An action invocation reached a successful terminal status.

InvocationFailed = 'invocation.failed'#: An action invocation reached a failed terminal status (Failed or Deadly).

ScheduleFired = 'schedule.fired'#

a trigger fires on a schedule by declaring a Schedule source, and the scheduler delivers this occurrence to that trigger alone.

Type:: A trigger’s own schedule reached one of its minutes. Not subscribable

SessionCreated = 'session.created'#: A session was created.

SessionFilesAdded = 'session.files_added'#: Files were added to a session.

SessionUpdated = 'session.updated'#: A session’s metadata or tags changed.

UploadCompleted = 'dataset.upload_completed'#: An upload transaction to a dataset completed (all of its files uploaded).

property cloudevents_type: str#

ai.roboto.file.uploaded.

Type:: This type as a CloudEvents type attribute
Return type:: str

classmethod from_cloudevents_type(value)#

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.
Return type:: PlatformEventType

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

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

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

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

roboto.domain.platform_events.SAMPLE_CHANGESET#

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

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

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

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

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

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

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

roboto.domain.platform_events.SAMPLE_TIME#: The instant every sample event and record is dated from, so samples are stable across calls.

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

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

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

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.ScheduleFired.

Parameters:: data (Any)

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

scheduled_for: datetime.datetime#: The scheduled minute (UTC) this occurrence stands for.

trigger_id: str#: The trigger whose schedule fired.

class roboto.domain.platform_events.SessionCreatedPayload(/, **data)#

Bases: pydantic.BaseModel

Payload for PlatformEventType.SessionCreated.

Parameters:: data (Any)

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

session_id: str#: The created session.

class roboto.domain.platform_events.SessionFilesAddedPayload(/, **data)#

Bases: pydantic.BaseModel

Payload for PlatformEventType.SessionFilesAdded.

Parameters:: data (Any)

file_ids: list[str]#: The added files.

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

session_id: str#: The session files were added to.

class roboto.domain.platform_events.SessionUpdatedPayload(/, **data)#

Bases: pydantic.BaseModel

Payload for PlatformEventType.SessionUpdated.

Parameters:: data (Any)

changeset: roboto.updates.MetadataChangeset#: The applied metadata/tag delta, served to conditions as the changed and tag roots.

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

session_id: str#: The session whose metadata changed.

roboto.domain.platform_events.TRIGGER_ROOT = 'trigger'#: Namespace root describing the trigger being evaluated. Reserved for the consumer; no event type may expose it as an entity root.

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

Bases: pydantic.BaseModel

Payload for PlatformEventType.UploadCompleted.

Parameters:: data (Any)

dataset_id: str#: Dataset the upload targeted.

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

transaction_id: str#: The completed upload transaction.

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

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:: catalog (PlatformEventCatalog)
Return type:: dict[str, Any]

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

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

roboto.domain.platform_events.sample_platform_event(event_type)#

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

Parameters:: event_type (roboto.domain.platform_events.events.PlatformEventType)
Return type:: roboto.domain.platform_events.events.PlatformEvent