roboto.domain.triggers.trigger#

Module Contents#

class roboto.domain.triggers.trigger.Trigger(record, roboto_client=None)#

A trigger: a firing source (event subscriptions or a schedule), an optional condition, and targets.

Triggers created through the older causes/for_each API (roboto.domain.actions.Trigger) are listed and loaded here too, projected onto this shape and marked engine == "v1"; they accept a narrower set of edits (see update()).

Examples

Start an agent whenever a dataset gains a ready tag:

>>> from roboto.domain.platform_events import OncePer, PlatformEventType
>>> from roboto.domain.triggers import StartAgentTarget, Trigger
>>> trigger = Trigger.create(
...     name="analyze-ready-datasets",
...     targets=[StartAgentTarget(target_id="analyze", agent_id="ag_abc123")],
...     events=[PlatformEventType.DatasetTagAdded],
...     once_per=OncePer.Occurrence,
... )

Post to Slack every Monday at 09:00 UTC:

>>> from roboto.domain.triggers import SendSlackMessageTarget
>>> weekly = Trigger.create(
...     name="weekly-status",
...     targets=[SendSlackMessageTarget(target_id="post", channel_id="C0123", text="Weekly check-in")],
...     schedule="0 9 * * 1",
... )
Parameters:
property condition: roboto.query.ConditionType | None#
Return type:

Optional[roboto.query.ConditionType]

classmethod create(name, targets, *, events=None, once_per=None, schedule=None, fires_on=None, condition=None, enabled=True, caller_org_id=None, roboto_client=None)#

Create a trigger in the caller’s org.

Parameters:

  • name (str) – Trigger name, unique within the org.

  • targets (list[roboto.domain.triggers.targets.TriggerTargetSpec]) – What to dispatch on a match. At least one.

  • events (Optional[list[roboto.domain.platform_events.PlatformEventType]]) – Platform event types to subscribe to (with once_per). At least one.

  • once_per (Optional[roboto.domain.platform_events.OncePer]) – What the trigger fires at most once per; must be legal for every subscribed event.

  • schedule (Optional[str]) – A cron expression (UTC) to fire on instead of events.

  • fires_on (Optional[roboto.domain.triggers.sources.TriggerSource]) – The firing source itself, as an alternative to the events/once_per or schedule shorthands.

  • condition (Optional[roboto.query.ConditionType]) – Optional predicate over the firing’s namespace.

  • enabled (bool) – Whether the trigger is active immediately.

  • caller_org_id (Optional[str]) – Org to create the trigger in. Defaults to the caller’s org.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses the default if not provided.

Returns:

The created trigger.

Raises:
Return type:

Trigger

delete()#

Delete this trigger. Idempotent.

Return type:

None

disable()#

Disable this trigger.

Return type:

Trigger

dispatches(limit=100)#

Yield this trigger’s dispatch history, newest first.

A dispatch is one attempt to run one target for one matched event; only matches are recorded, so an empty history means the trigger never fired.

Parameters:

limit (int) – Page size for the underlying requests.

Return type:

collections.abc.Generator[roboto.domain.triggers.dispatch.TriggerDispatchRecord, None, None]

dry_run(event=None, dataset_id=None, file_id=None, invocation_id=None, session_id=None, event_id=None, event_type=None, scheduled_for=None)#

Ask “would this trigger fire?” without dispatching anything.

Provide either a full event or exactly one entity reference, from which the server synthesizes an event (event_type optionally picks which kind). A schedule-fired trigger takes no reference: pass scheduled_for to pick the minute, or nothing for the schedule’s next occurrence. The response is the evaluator’s gate-by-gate trace: subscribed, enabled, condition (with per-leaf actual values), target prefilter, already fired.

Examples

>>> trace = trigger.dry_run(dataset_id="ds_abc123")
>>> print(trace.verdict)
Parameters:
Return type:

roboto.domain.triggers.dry_run.TriggerDryRunResponse

enable()#

Enable this trigger.

Return type:

Trigger

property enabled: bool#
Return type:

bool

property engine: str#

"v2", or "v1" for a trigger created through the older causes/for_each API.

Type:

Which trigger shape this trigger was created in

Return type:

str

property events: List[roboto.domain.platform_events.PlatformEventType] | None#

The subscribed event types, or None for a schedule-fired trigger.

Return type:

Optional[List[roboto.domain.platform_events.PlatformEventType]]

property fires_on: roboto.domain.triggers.sources.TriggerSource#

an event subscription or a schedule.

Type:

What makes the trigger fire

Return type:

roboto.domain.triggers.sources.TriggerSource

classmethod from_id(trigger_id, roboto_client=None)#

Load the trigger with the given id, whichever API created it.

Parameters:
Return type:

Trigger

classmethod from_name(name, owner_org_id=None, roboto_client=None)#

Load the trigger with the given name, whichever API created it. Names are unique within an org.

Parameters:
Return type:

Trigger

classmethod list(owner_org_id=None, roboto_client=None)#

Yield every trigger in the org, event-fired and scheduled, newest first, whichever API created it.

Parameters:
Return type:

collections.abc.Generator[Trigger, None, None]

property name: str#
Return type:

str

property once_per: roboto.domain.platform_events.OncePer#

What the trigger fires at most once per; always occurrence (one firing per minute) for a schedule.

Return type:

roboto.domain.platform_events.OncePer

property org_id: str#
Return type:

str

classmethod platform_event_samples(roboto_client=None)#

Fetch a realistic sample of every event type, dereferenced.

Each sample carries the event envelope, every namespace root the type exposes with a full record under it, and the root.path list a condition field or {{ }} placeholder may name. Use it to see what a template will resolve to before writing one.

Examples

>>> samples = Trigger.platform_event_samples()
>>> samples[PlatformEventType.FileUploaded].paths[:3]
['envelope.id', 'envelope.type', 'envelope.time']
Parameters:

roboto_client (Optional[roboto.http.RobotoClient])

Return type:

dict[roboto.domain.platform_events.PlatformEventType, roboto.domain.triggers.samples.PlatformEventSample]

property record: roboto.domain.triggers.record.TriggerRecord#
Return type:

roboto.domain.triggers.record.TriggerRecord

property schedule: str | None#

The cron expression (UTC) the trigger fires on, or None for an event-fired trigger.

Return type:

Optional[str]

set_enabled(enabled)#

Enable or disable this trigger.

Parameters:

enabled (bool)

Return type:

Trigger

property targets: List[roboto.domain.triggers.targets.TriggerTargetSpec]#
Return type:

List[roboto.domain.triggers.targets.TriggerTargetSpec]

to_dict()#

Return this trigger’s record as a JSON-able dict.

Return type:

dict[str, Any]

property trigger_id: str#
Return type:

str

update(*, fires_on=NotSet, events=NotSet, once_per=NotSet, schedule=NotSet, condition=NotSet, targets=NotSet, enabled=NotSet)#

Apply a partial update to this trigger and refresh this instance.

Only provided fields change; condition=None clears the condition. The firing source is replaced whole: pass fires_on, or the events / once_per / schedule shorthands, which are merged over the current source before being sent. A trigger with engine == "v1" accepts only edits its older shape can express — a single invoke-action target, event types that map onto it, once_per of file or dataset — and rejects the rest with a message naming what it cannot store.

Parameters:
Return type:

Trigger

On this page