roboto.domain.triggers.trigger
Module Contents
Trigger
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()).
Usage
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
roboto_client Optional[roboto.Properties
Trigger.condition
Trigger.create()
Create a trigger in the caller’s org.
Parameters
name strTrigger name, unique within the org.
What to dispatch on a match. At least one.
events Optional[list[roboto.Platform event types to subscribe to (with once_per). At least one.
once_per Optional[roboto.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.The firing source itself, as an alternative to the events/once_per or schedule shorthands.
condition Optional[roboto.Optional predicate over the firing’s namespace.
enabled boolWhether 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.Roboto client instance. Uses the default if not provided.
Returns
The created trigger.
Raises
A cross-field validation rule fails (e.g. the condition references a root not exposed by every subscribed event).
The name is already taken in the org.
Trigger.delete()
Delete this trigger. Idempotent.
Return type
Trigger.disable()
Trigger.dispatches()
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 intPage size for the underlying requests.
Return type
Trigger.dry_run()
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.
Usage
trace = trigger.dry_run(dataset_id="ds_abc123")
print(trace.verdict)Parameters
event Optional[roboto.dataset_id Optional[str]file_id Optional[str]invocation_id Optional[str]event_id Optional[str]event_type Optional[roboto.scheduled_for Optional[datetime.Trigger.enable()
Properties
Trigger.engine
Which trigger shape this trigger was created in: "v2", or "v1" for a trigger created through the older causes/for_each API.
Trigger.events
The subscribed event types, or None for a schedule-fired trigger.
Trigger.fires_on
What makes the trigger fire: an event subscription or a schedule.
Trigger.from_id()
Load the trigger with the given id, whichever API created it.
Parameters
trigger_id strroboto_client Optional[roboto.Return type
Trigger.from_name()
Load the trigger with the given name, whichever API created it. Names are unique within an org.
Parameters
Return type
Trigger.list()
Yield every trigger in the org, event-fired and scheduled, newest first, whichever API created it.
Parameters
owner_org_id Optional[str]roboto_client Optional[roboto.Return type
Properties
Trigger.once_per
What the trigger fires at most once per; always occurrence (one firing per minute) for a schedule.
Trigger.platform_event_samples()
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.
Usage
samples = Trigger.platform_event_samples()
samples[PlatformEventType.FileUploaded].paths[:3]
# ['envelope.id', 'envelope.type', 'envelope.time']Parameters
roboto_client Optional[roboto.Properties
Trigger.record
Trigger.schedule
The cron expression (UTC) the trigger fires on, or None for an event-fired trigger.
Trigger.set_enabled()
Enable or disable this trigger.
Parameters
enabled boolReturn type
Properties
Trigger.targets
Trigger.to_dict()
Return this trigger’s record as a JSON-able dict.
Return type
Properties
Trigger.update()
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
fires_on Union[roboto.events Union[List[roboto.once_per Union[roboto.schedule Union[str, roboto.condition Optional[Union[roboto.targets Union[List[roboto.enabled Union[bool, roboto.Return type