trigger

trigger

Classes for managing Domo trigger configurations (dataflow execution triggers)

This module extends the DomoSchedule system to handle complex trigger scenarios: - Multiple triggers per entity (vs single schedule) - Dataset-based triggers (in addition to time-based) - Conditional execution logic - Multiple event types per trigger

DomoTriggerEvent_Schedule leverages the DomoSchedule factory to parse schedule configurations, ensuring consistency with how DomoStream and other entities handle time-based execution.

Architecture

DomoTriggerSettings (container) └── DomoTrigger (individual trigger) ├── DomoTriggerEvent_Schedule --> DomoSchedule_Base (reuses schedule parsing) ├── DomoTriggerEvent_DatasetUpdated └── DomoTriggerCondition

DomoTrigger dataclass

DomoTrigger(
    trigger_id: int,
    title: str = None,
    trigger_events: list[DomoTriggerEvent_Base] = list(),
    trigger_conditions: list[DomoTriggerCondition] = list(),
    raw: dict[str, Any] = dict(),
)

Bases: DomoBase

A single trigger configuration with events and conditions

export_as_dict

export_as_dict(
    override_fn: Callable | None = None,
) -> dict[str, Any]

Export trigger as unified dictionary format

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Optional function to override export logic	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Unified trigger dictionary with readable format

Source code in src/crew_dcs/classes/subentity/trigger.py

def export_as_dict(self, override_fn: Callable | None = None) -> dict[str, Any]:
    """Export trigger as unified dictionary format

    Args:
        override_fn: Optional function to override export logic

    Returns:
        dict: Unified trigger dictionary with readable format
    """
    if override_fn:
        return override_fn(self)

    return {
        "triggerId": self.trigger_id,
        "title": self.title,
        "triggerEvents": [
            (
                event.export_as_dict()
                if hasattr(event, "export_as_dict")
                else event.to_dict()
            )
            for event in self.trigger_events
        ],
        "triggerConditions": [
            (
                condition.export_as_dict()
                if hasattr(condition, "export_as_dict")
                else condition.to_dict()
            )
            for condition in self.trigger_conditions
        ],
        "humanReadable": self.get_human_readable_description(),
    }

from_dict classmethod

from_dict(obj: dict[str, Any]) -> DomoTrigger

Create DomoTrigger from dictionary

Source code in src/crew_dcs/classes/subentity/trigger.py

@classmethod
def from_dict(cls, obj: dict[str, Any]) -> "DomoTrigger":
    """Create DomoTrigger from dictionary"""
    trigger_events = [
        DomoTriggerEvent_Base.from_dict(event)
        for event in obj.get("triggerEvents", [])
    ]

    trigger_conditions = [
        DomoTriggerCondition.from_dict(condition)
        for condition in obj.get("triggerConditions", [])
    ]

    return cls(
        trigger_id=obj.get("triggerId"),
        title=obj.get("title"),
        trigger_events=trigger_events,
        trigger_conditions=trigger_conditions,
        raw=obj,
    )

get_dataset_events

get_dataset_events() -> (
    list[DomoTriggerEvent_DatasetUpdated]
)

Get all dataset update events from this trigger

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_dataset_events(self) -> list[DomoTriggerEvent_DatasetUpdated]:
    """Get all dataset update events from this trigger"""
    return [
        event
        for event in self.trigger_events
        if isinstance(event, DomoTriggerEvent_DatasetUpdated)
    ]

get_human_readable_description

get_human_readable_description() -> str

Get human-readable description of the trigger

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_human_readable_description(self) -> str:
    """Get human-readable description of the trigger"""
    parts = []

    if self.title:
        parts.append(f"'{self.title}'")

    # Describe events
    if self.trigger_events:
        event_descriptions = [
            event.get_human_readable_description() for event in self.trigger_events
        ]
        parts.append(f"Events: {'; '.join(event_descriptions)}")

    # Describe conditions
    if self.trigger_conditions:
        condition_descriptions = [
            condition.get_human_readable_description()
            for condition in self.trigger_conditions
        ]
        parts.append(f"Conditions: {'; '.join(condition_descriptions)}")

    return " | ".join(parts) if parts else f"Trigger {self.trigger_id}"

get_schedule_events

get_schedule_events() -> list[DomoTriggerEvent_Schedule]

Get all schedule events from this trigger

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_schedule_events(self) -> list[DomoTriggerEvent_Schedule]:
    """Get all schedule events from this trigger"""
    return [
        event
        for event in self.trigger_events
        if isinstance(event, DomoTriggerEvent_Schedule)
    ]

has_dataset_event

has_dataset_event() -> bool

Check if this trigger has a dataset update event

Source code in src/crew_dcs/classes/subentity/trigger.py

def has_dataset_event(self) -> bool:
    """Check if this trigger has a dataset update event"""
    return any(
        isinstance(event, DomoTriggerEvent_DatasetUpdated)
        for event in self.trigger_events
    )

has_schedule_event

has_schedule_event() -> bool

Check if this trigger has a schedule event

Source code in src/crew_dcs/classes/subentity/trigger.py

def has_schedule_event(self) -> bool:
    """Check if this trigger has a schedule event"""
    return any(
        isinstance(event, DomoTriggerEvent_Schedule)
        for event in self.trigger_events
    )

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary format

Source code in src/crew_dcs/classes/subentity/trigger.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary format"""
    return {
        "triggerId": self.trigger_id,
        "title": self.title,
        "triggerEvents": [event.to_dict() for event in self.trigger_events],
        "triggerConditions": [
            condition.to_dict() for condition in self.trigger_conditions
        ],
    }

DomoTriggerCondition dataclass

DomoTriggerCondition(
    condition_type: str = None,
    parameters: dict[str, Any] = dict(),
    raw: dict[str, Any] = dict(),
)

Bases: DomoBase

Condition that must be met for trigger to fire

export_as_dict

export_as_dict(
    override_fn: Callable | None = None,
) -> dict[str, Any]

Export condition as unified dictionary format

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Optional function to override export logic	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Unified condition dictionary with readable format

Source code in src/crew_dcs/classes/subentity/trigger.py

def export_as_dict(self, override_fn: Callable | None = None) -> dict[str, Any]:
    """Export condition as unified dictionary format

    Args:
        override_fn: Optional function to override export logic

    Returns:
        dict: Unified condition dictionary with readable format
    """
    if override_fn:
        return override_fn(self)

    return {
        "type": self.condition_type,
        "parameters": self.parameters,
        "humanReadable": self.get_human_readable_description(),
    }

get_human_readable_description

get_human_readable_description() -> str

Get human-readable description of the condition

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_human_readable_description(self) -> str:
    """Get human-readable description of the condition"""
    if not self.condition_type:
        return "No conditions"
    return f"Condition: {self.condition_type}"

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary format

Source code in src/crew_dcs/classes/subentity/trigger.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary format"""
    return {
        "type": self.condition_type,
        "parameters": self.parameters,
        **{k: v for k, v in self.raw.items() if k not in ["type", "parameters"]},
    }

DomoTriggerEvent_Base dataclass

DomoTriggerEvent_Base(
    event_type: TriggerEventType,
    raw: dict[str, Any] = dict(),
)

Bases: DomoBase, ABC

Base class for trigger events

from_dict classmethod

from_dict(obj: dict[str, Any]) -> DomoTriggerEvent_Base

Factory method to create appropriate trigger event subclass

Source code in src/crew_dcs/classes/subentity/trigger.py

@classmethod
def from_dict(cls, obj: dict[str, Any]) -> "DomoTriggerEvent_Base":
    """Factory method to create appropriate trigger event subclass"""
    event_type = obj.get("type", "").upper()

    if event_type == "DATASET_UPDATED":
        return DomoTriggerEvent_DatasetUpdated.from_dict(obj)
    if event_type == "SCHEDULE":
        return DomoTriggerEvent_Schedule.from_dict(obj)
    # Return a generic trigger event for unknown types
    return DomoTriggerEvent_Generic.from_dict(obj)

get_human_readable_description abstractmethod

get_human_readable_description() -> str

Get human-readable description of the trigger event

Source code in src/crew_dcs/classes/subentity/trigger.py

@abstractmethod
def get_human_readable_description(self) -> str:
    """Get human-readable description of the trigger event"""

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary format

Source code in src/crew_dcs/classes/subentity/trigger.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary format"""
    return {
        "type": self.event_type.value,
        **self.raw,
    }

DomoTriggerEvent_DatasetUpdated dataclass

DomoTriggerEvent_DatasetUpdated(
    event_type: TriggerEventType,
    raw: dict[str, Any] = dict(),
    dataset_id: str = None,
    trigger_on_data_changed: bool = False,
)

Bases: DomoTriggerEvent_Base

Trigger event for dataset updates

export_as_dict

export_as_dict(
    override_fn: Callable | None = None,
) -> dict[str, Any]

Export event as unified dictionary format

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Optional function to override export logic	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Unified event dictionary with readable format

Source code in src/crew_dcs/classes/subentity/trigger.py

def export_as_dict(self, override_fn: Callable | None = None) -> dict[str, Any]:
    """Export event as unified dictionary format

    Args:
        override_fn: Optional function to override export logic

    Returns:
        dict: Unified event dictionary with readable format
    """
    if override_fn:
        return override_fn(self)

    return {
        "type": self.event_type.value,
        "datasetId": self.dataset_id,
        "triggerOnDataChanged": self.trigger_on_data_changed,
        "humanReadable": self.get_human_readable_description(),
    }

get_human_readable_description

get_human_readable_description() -> str

Get human-readable description

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_human_readable_description(self) -> str:
    """Get human-readable description"""
    change_text = (
        "when data changes" if self.trigger_on_data_changed else "on every update"
    )
    return f"Dataset {self.dataset_id} updated ({change_text})"

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary format

Source code in src/crew_dcs/classes/subentity/trigger.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary format"""
    return {
        "type": self.event_type.value,
        "datasetId": self.dataset_id,
        "triggerOnDataChanged": self.trigger_on_data_changed,
    }

DomoTriggerEvent_Generic dataclass

DomoTriggerEvent_Generic(
    event_type: TriggerEventType,
    raw: dict[str, Any] = dict(),
)

Bases: DomoTriggerEvent_Base

Generic trigger event for unknown/future types

DomoTriggerEvent_Schedule dataclass

DomoTriggerEvent_Schedule(
    event_type: TriggerEventType,
    raw: dict[str, Any] = dict(),
    schedule_id: str | None = None,
    schedule: DomoSchedule_Base | None = None,
)

Bases: DomoTriggerEvent_Base

Trigger event for scheduled execution

Uses DomoSchedule factory to parse and interpret schedule configurations, ensuring consistency with Stream and other schedule implementations.

export_as_dict

export_as_dict(
    override_fn: Callable | None = None,
) -> dict[str, Any]

Export event as unified dictionary format

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Optional function to override export logic	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Unified event dictionary with readable format

Source code in src/crew_dcs/classes/subentity/trigger.py

def export_as_dict(self, override_fn: Callable | None = None) -> dict[str, Any]:
    """Export event as unified dictionary format

    Args:
        override_fn: Optional function to override export logic

    Returns:
        dict: Unified event dictionary with readable format
    """
    if override_fn:
        return override_fn(self)

    result = {
        "type": self.event_type.value,
        "humanReadable": self.get_human_readable_description(),
    }

    if self.schedule_id:
        result["scheduleId"] = self.schedule_id

    # Include schedule details using DomoSchedule's export_as_dict
    if self.schedule:
        result["schedule"] = self.schedule.export_as_dict()

    return result

from_dict classmethod

from_dict(obj: dict[str, Any]) -> DomoTriggerEvent_Schedule

Create schedule event from trigger event dict

Leverages DomoSchedule factory to determine the appropriate schedule type (Simple, Cron, or Advanced) based on the schedule data structure.

Source code in src/crew_dcs/classes/subentity/trigger.py

@classmethod
def from_dict(cls, obj: dict[str, Any]) -> "DomoTriggerEvent_Schedule":
    """Create schedule event from trigger event dict

    Leverages DomoSchedule factory to determine the appropriate schedule type
    (Simple, Cron, or Advanced) based on the schedule data structure.
    """
    schedule_id = obj.get("id")
    schedule_data = obj.get("schedule")

    # Parse schedule if present using DomoSchedule factory
    schedule = None
    if schedule_data:
        # Build cron expression for DomoSchedule
        cron_expr = cls._build_cron_expression(schedule_data)

        # Create schedule dict for DomoSchedule factory
        schedule_dict = {
            "scheduleExpression": cron_expr,
            "timezone": schedule_data.get("timezone"),
        }

        # Use DomoSchedule factory to determine appropriate schedule type
        # This ensures consistency with how Streams handle schedules
        schedule_class = DomoSchedule.determine_schedule_type(schedule_dict)
        schedule = schedule_class.from_dict(schedule_dict, parent=None)

        # Store raw schedule data for round-trip serialization
        schedule.raw["raw_schedule"] = schedule_data

    return cls(schedule_id=schedule_id, schedule=schedule, raw=obj)

get_human_readable_description

get_human_readable_description() -> str

Get human-readable description

For schedule triggers, returns the cron expression when available, otherwise returns a human-friendly description.

Returns:

Name	Type	Description
str	str	Human-readable description like "0 47 12 ? * MON *" or "Schedule: Daily at 12:00"

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_human_readable_description(self) -> str:
    """Get human-readable description

    For schedule triggers, returns the cron expression when available,
    otherwise returns a human-friendly description.

    Returns:
        str: Human-readable description like "0 47 12 ? * MON *" or "Schedule: Daily at 12:00"
    """
    if self.schedule:
        # For custom cron schedules, show the expression instead of generic "Custom schedule"
        from .schedule import DomoCronSchedule, ScheduleFrequencyEnum

        if (
            self.schedule.frequency == ScheduleFrequencyEnum.CUSTOM_CRON
            and isinstance(self.schedule, DomoCronSchedule)
            and self.schedule.schedule_expression
        ):
            return self.schedule.schedule_expression

        return self.schedule.get_human_readable_schedule()
    return "Scheduled trigger"

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary format

Source code in src/crew_dcs/classes/subentity/trigger.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary format"""
    result = {
        "type": self.event_type.value,
    }

    if self.schedule_id:
        result["id"] = self.schedule_id

    if self.schedule and hasattr(self.schedule, "raw"):
        raw_schedule = self.schedule.raw.get("raw_schedule")
        if raw_schedule:
            result["schedule"] = raw_schedule

    return result

DomoTriggerSettings dataclass

DomoTriggerSettings(
    triggers: list[DomoTrigger] = list(),
    zone_id: str = "UTC",
    locale: str = "en_US",
    raw: dict[str, Any] = dict(),
    parent: Any = None,
)

Bases: DomoBase

Manager for trigger settings (collection of triggers)

export_as_dict

export_as_dict(
    override_fn: Callable | None = None,
) -> dict[str, Any]

Export trigger settings as unified dictionary format

Calls export_as_dict on each trigger to get detailed, readable format.

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Optional function to override export logic	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Unified trigger settings dictionary with readable format

Source code in src/crew_dcs/classes/subentity/trigger.py

def export_as_dict(self, override_fn: Callable | None = None) -> dict[str, Any]:
    """Export trigger settings as unified dictionary format

    Calls export_as_dict on each trigger to get detailed, readable format.

    Args:
        override_fn: Optional function to override export logic

    Returns:
        dict: Unified trigger settings dictionary with readable format
    """
    if override_fn:
        return override_fn(self)

    return {
        "triggers": [trigger.export_as_dict() for trigger in self.triggers],
        "zoneId": self.zone_id,
        "locale": self.locale,
        "summary": self.get_human_readable_summary(),
        "stats": {
            "totalTriggers": len(self.triggers),
            "scheduleTriggers": len(self.get_schedule_triggers()),
            "datasetTriggers": len(self.get_dataset_triggers()),
        },
    }

from_dict classmethod

from_dict(
    obj: dict[str, Any], parent: Any = None
) -> DomoTriggerSettings

Create DomoTriggerSettings from dictionary

Source code in src/crew_dcs/classes/subentity/trigger.py

@classmethod
def from_dict(
    cls, obj: dict[str, Any], parent: Any = None
) -> "DomoTriggerSettings":
    """Create DomoTriggerSettings from dictionary"""
    triggers = [
        DomoTrigger.from_dict(trigger) for trigger in obj.get("triggers", [])
    ]

    return cls(
        triggers=triggers,
        zone_id=obj.get("zoneId", "UTC"),
        locale=obj.get("locale", "en_US"),
        parent=parent,
        raw=obj,
    )

from_parent classmethod

from_parent(
    parent: Any, obj: dict[str, Any] | None = None
) -> DomoTriggerSettings

Create from parent entity

Source code in src/crew_dcs/classes/subentity/trigger.py

@classmethod
def from_parent(
    cls, parent: Any, obj: dict[str, Any] | None = None
) -> "DomoTriggerSettings":
    """Create from parent entity"""
    if obj is None:
        obj = {}

    return cls.from_dict(obj, parent=parent)

get_dataset_triggers

get_dataset_triggers() -> list[DomoTrigger]

Get all triggers that have dataset update events

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_dataset_triggers(self) -> list[DomoTrigger]:
    """Get all triggers that have dataset update events"""
    return [trigger for trigger in self.triggers if trigger.has_dataset_event()]

get_human_readable_summary

get_human_readable_summary() -> str

Get human-readable summary of all triggers

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_human_readable_summary(self) -> str:
    """Get human-readable summary of all triggers"""
    if not self.triggers:
        return "No triggers configured"

    parts = []
    for trigger in self.triggers:
        parts.append(f"  • {trigger.get_human_readable_description()}")

    return f"Triggers ({len(self.triggers)}):\n" + "\n".join(parts)

get_schedule_triggers

get_schedule_triggers() -> list[DomoTrigger]

Get all triggers that have schedule events

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_schedule_triggers(self) -> list[DomoTrigger]:
    """Get all triggers that have schedule events"""
    return [trigger for trigger in self.triggers if trigger.has_schedule_event()]

get_trigger_by_id

get_trigger_by_id(trigger_id: int) -> DomoTrigger | None

Get a specific trigger by ID

Source code in src/crew_dcs/classes/subentity/trigger.py

def get_trigger_by_id(self, trigger_id: int) -> DomoTrigger | None:
    """Get a specific trigger by ID"""
    for trigger in self.triggers:
        if trigger.trigger_id == trigger_id:
            return trigger
    return None

has_any_dataset_triggers

has_any_dataset_triggers() -> bool

Check if any trigger has dataset update events

Source code in src/crew_dcs/classes/subentity/trigger.py

def has_any_dataset_triggers(self) -> bool:
    """Check if any trigger has dataset update events"""
    return len(self.get_dataset_triggers()) > 0

has_any_schedules

has_any_schedules() -> bool

Check if any trigger has a schedule event

Source code in src/crew_dcs/classes/subentity/trigger.py

def has_any_schedules(self) -> bool:
    """Check if any trigger has a schedule event"""
    return len(self.get_schedule_triggers()) > 0

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary format

Source code in src/crew_dcs/classes/subentity/trigger.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary format"""
    return {
        "triggers": [trigger.to_dict() for trigger in self.triggers],
        "zoneId": self.zone_id,
        "locale": self.locale,
    }

TriggerEventType

Bases: DomoEnumMixin, Enum

Types of trigger events