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, in this shape.

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 and its dispatch history.

Deleting a trigger that is already deleted is a no-op.

Raises:
Return type:

None

Examples

>>> trigger = Trigger.from_name("weekly-status")
>>> trigger.delete()
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, 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#

Which engine evaluates this trigger; see TriggerRecord.engine.

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#

What makes the trigger fire, an event subscription or a schedule.

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.

Parameters:
Raises:

  • ValueError – The trigger fires on a schedule and events is given without once_per.

  • RobotoNotFoundException – The trigger no longer exists, or a target names an action that does not.

  • RobotoInvalidRequestException – The updated trigger fails validation, a target names an agent or Slack channel the org does not have, or the trigger is managed (see TriggerRecord.managed_by): a trigger an ingestion rule owns changes only through its rule, and one Roboto manages can only be enabled or disabled.

  • RobotoUnauthorizedException – The caller is not a member of the trigger’s org, or a target names an action the org may not run or a dataset the caller may not upload to.

  • RobotoLimitExceededException – Enabling the trigger would put the org over its limit of active triggers, or a target’s timeout exceeds the org’s limit for action timeouts.

  • RobotoConflictException – The trigger kept changing while the update was applied; retry it.

Return type:

Trigger

Examples

Disable a trigger:

>>> trigger = Trigger.from_name("analyze-ready-datasets")
>>> trigger.update(enabled=False)

Fire on dataset creation instead, keeping its once_per:

>>> from roboto.domain.platform_events import PlatformEventType
>>> trigger.update(events=[PlatformEventType.DatasetCreated])

Clear its condition:

>>> trigger.update(condition=None)
On this page