roboto.domain.triggers.sources#

What makes a trigger fire: a platform event subscription or a schedule.

A trigger has exactly one firing source. Both kinds feed the same condition and the same targets, and a target cannot tell which one fired it beyond what the event namespace exposes.

Module Contents#

class roboto.domain.triggers.sources.EventSubscription(/, **data)#

Bases: _SourceBase

Fire when any of the subscribed platform events occurs.

Carries once_per because it only means something here: it names what the event is about that the trigger fires at most once for, and a schedule has no such subject.

Parameters:: data (Any)

events: list[roboto.domain.platform_events.PlatformEventType]#: Event types the trigger subscribes to. At least one; a condition may only reference namespace roots exposed by every subscribed event.

fires_for(event_type)#

Return whether an event of event_type is one this subscription fires for.

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

once_per: roboto.domain.platform_events.OncePer#

the occurrence, or an entity the event names. Must be legal for every subscribed event type.

Type:: What the trigger fires at most once per

type: Literal[TriggerSourceType]#: Discriminator for TriggerSource.

class roboto.domain.triggers.sources.Schedule(/, **data)#

Bases: _SourceBase

Fire on a cron schedule, in UTC.

Each scheduled minute fires the trigger once (once_per='occurrence' over the schedule tick). A missed minute is skipped rather than replayed later, so a delayed schedule never floods targets with backdated firings. Cron expressions are evaluated in UTC; a schedule cannot name a time zone.

Parameters:: data (Any)

cron: str#: A five-field cron expression (minute hour day-of-month month day-of-week), evaluated in UTC — e.g. 0 9 * * 1 for 09:00 UTC every Monday.

fires_for(event_type)#

Return whether event_type is the schedule occurrence this source fires for.

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

property once_per: roboto.domain.platform_events.OncePer#

A schedule always fires once per scheduled minute.

Return type:: roboto.domain.platform_events.OncePer

type: Literal[TriggerSourceType]#: Discriminator for TriggerSource.

roboto.domain.triggers.sources.TriggerSource#: A trigger’s firing source, discriminated on type.

class roboto.domain.triggers.sources.TriggerSourceType#

Bases: roboto.compat.StrEnum

Discriminator for TriggerSource.

Events = 'events'#: The trigger fires when a subscribed platform event occurs.

Schedule = 'schedule'#: The trigger fires on a cron schedule.