trigger

trigger

DomoWorkflow_Trigger_Timer dataclass

DomoWorkflow_Trigger_Timer(
    id: str,
    auth: DomoAuth,
    name: str,
    model_id: str,
    model_version: str,
    cron: str,
    time_zone: str = "UTC",
    entity_id: str | None = None,
    active: bool = True,
    created_by: str | None = None,
    created_dt: datetime | None = None,
    updated_by: str | None = None,
    updated_dt: datetime | None = None,
    deleted_dt: datetime | None = None,
    raw: dict = dict(),
)

A TIMER trigger that fires a workflow model/version on a cron schedule.

Timer triggers are NOT part of a workflow's canvas (designElements/savedJson, the thing DomoWorkflow_Tile subclasses represent) — they are a separate resource created via POST /api/workflow/v2/triggers, scoped to the creating user (entityType="USER", entityId=; Domo rejects any other entityType/entityId combo for TIMER triggers with "Timer triggers can only be created for the current user"). Reverse-engineered against domo-community since no shipped workflow there used a non-manual trigger.

create async classmethod

create(
    auth: DomoAuth,
    name: str,
    model_id: str,
    model_version: str,
    cron: str,
    time_zone: str = "UTC",
    entity_id: str | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow_Trigger_Timer

Create a TIMER trigger that fires model_id@model_version on a cron schedule.

cron is a Quartz-style 6-field expression: "sec min hour day month dow" (e.g. "0 0 2 * * ?" = daily at 02:00). entity_id defaults to the authenticated user — Domo requires TIMER triggers be scoped to their creator.

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

@classmethod
async def create(
    cls,
    auth: DomoAuth,
    name: str,
    model_id: str,
    model_version: str,
    cron: str,
    time_zone: str = "UTC",
    entity_id: str | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow_Trigger_Timer:
    """Create a TIMER trigger that fires `model_id`@`model_version` on a cron schedule.

    `cron` is a Quartz-style 6-field expression: "sec min hour day month dow"
    (e.g. "0 0 2 * * ?" = daily at 02:00). `entity_id` defaults to the
    authenticated user — Domo requires TIMER triggers be scoped to their creator.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if entity_id is None:
        if not auth.user_id:
            await auth.who_am_i()
        entity_id = auth.user_id

    res = await workflow_routes.create_workflow_trigger(
        auth=auth,
        name=name,
        trigger_type="TIMER",
        model_id=model_id,
        model_version=model_version,
        entity_type="USER",
        entity_id=entity_id,
        metadata={
            "type": "TIMER",
            "cron": cron,
            "timeZone": time_zone,
            "interval": "Every Day",
            "minuteDuration": None,
            "activeHours": None,
            "showActiveHours": True,
            "startHour": 0,
            "dayOfWeek": None,
            "dayOfMonth": None,
        },
        context=context,
    )

    return cls.from_dict(auth=auth, obj=res.response)

create_daily async classmethod

create_daily(
    auth: DomoAuth,
    name: str,
    model_id: str,
    model_version: str,
    hour: int,
    minute: int = 0,
    time_zone: str = "UTC",
    entity_id: str | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow_Trigger_Timer

Convenience wrapper for the common "run once a day at HH:MM" schedule.

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

@classmethod
async def create_daily(
    cls,
    auth: DomoAuth,
    name: str,
    model_id: str,
    model_version: str,
    hour: int,
    minute: int = 0,
    time_zone: str = "UTC",
    entity_id: str | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow_Trigger_Timer:
    """Convenience wrapper for the common "run once a day at HH:MM" schedule."""
    cron = f"0 {minute} {hour} * * ?"
    return await cls.create(
        auth=auth,
        name=name,
        model_id=model_id,
        model_version=model_version,
        cron=cron,
        time_zone=time_zone,
        entity_id=entity_id,
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

delete async

delete(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> bool

Soft-delete this trigger (stops future firings).

Parameters:

Name	Type	Description	Default
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
bool	True when Domo accepted the delete.

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

async def delete(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> bool:
    """Soft-delete this trigger (stops future firings).

    Args:
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        True when Domo accepted the delete.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await workflow_routes.delete_workflow_trigger(
        auth=self.auth,
        trigger_id=self.id,
        context=context,
    )

    return res.is_success

from_dict classmethod

from_dict(
    auth: DomoAuth, obj: dict[str, Any]
) -> DomoWorkflow_Trigger_Timer

Build a timer trigger from a /v2/triggers API payload.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
obj	dict[str, Any]	Raw trigger dict.	required

Returns:

Type	Description
DomoWorkflow_Trigger_Timer	The populated trigger.

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

@classmethod
def from_dict(
    cls, auth: DomoAuth, obj: dict[str, Any]
) -> DomoWorkflow_Trigger_Timer:
    """Build a timer trigger from a ``/v2/triggers`` API payload.

    Args:
        auth: Authenticated Domo session.
        obj: Raw trigger dict.

    Returns:
        The populated trigger.
    """
    metadata = obj.get("metadata") or {}
    return cls(
        id=obj.get("id", ""),
        auth=auth,
        name=obj.get("name", ""),
        model_id=obj.get("modelId", ""),
        model_version=obj.get("modelVersion", ""),
        cron=metadata.get("cron", ""),
        time_zone=metadata.get("timeZone", "UTC"),
        entity_id=obj.get("entityId"),
        active=obj.get("active", True),
        created_by=obj.get("createdBy"),
        created_dt=convert_string_to_datetime(obj.get("createdAt")),
        updated_by=obj.get("updatedBy"),
        updated_dt=convert_string_to_datetime(obj.get("updatedAt")),
        deleted_dt=convert_string_to_datetime(obj.get("deletedAt")),
        raw=obj,
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    trigger_id: str,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow_Trigger_Timer

Fetch a trigger by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
trigger_id	str	Trigger UUID.	required
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
DomoWorkflow_Trigger_Timer	The trigger.

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

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    trigger_id: str,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow_Trigger_Timer:
    """Fetch a trigger by ID.

    Args:
        auth: Authenticated Domo session.
        trigger_id: Trigger UUID.
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The trigger.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await workflow_routes.get_workflow_trigger_by_id(
        auth=auth,
        trigger_id=trigger_id,
        context=context,
    )

    return cls.from_dict(auth=auth, obj=res.response)

list_by_entity async classmethod

list_by_entity(
    auth: DomoAuth,
    entity_type: str,
    entity_id: str,
    entity_versions: str | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List triggers attached to an entity.

Returns the raw response: the listing spans every trigger type, not only TIMER, so it is not converted into DomoWorkflow_Trigger_Timer.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
entity_type	str	Entity type path segment.	required
entity_id	str	Entity UUID.	required
entity_versions	str | None	Optional entity versions to filter by.	None
return_raw	bool	If True, return the response without raising on failure.	False
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

Raises:

Type	Description
Workflow_GET_Error	If the response is not successful and return_raw is False.

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

@classmethod
async def list_by_entity(
    cls,
    auth: DomoAuth,
    entity_type: str,
    entity_id: str,
    entity_versions: str | None = None,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List triggers attached to an entity.

    Returns the raw response: the listing spans every trigger type, not
    only TIMER, so it is not converted into ``DomoWorkflow_Trigger_Timer``.

    Args:
        auth: Authenticated Domo session.
        entity_type: Entity type path segment.
        entity_id: Entity UUID.
        entity_versions: Optional entity versions to filter by.
        return_raw: If True, return the response without raising on failure.
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and
            ``return_raw`` is False.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.list_workflow_triggers_by_entity(
        auth=auth,
        entity_type=entity_type,
        entity_id=entity_id,
        entity_versions=entity_versions,
        return_raw=return_raw,
        context=context,
    )

list_by_model async classmethod

list_by_model(
    auth: DomoAuth,
    workflow_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List the triggers of a workflow model.

Returns the raw response: the listing spans every trigger type, not only TIMER, so it is not converted into DomoWorkflow_Trigger_Timer.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
workflow_id	str	Workflow UUID.	required
return_raw	bool	If True, return the response without raising on failure.	False
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

Raises:

Type	Description
Workflow_GET_Error	If the response is not successful and return_raw is False.

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

@classmethod
async def list_by_model(
    cls,
    auth: DomoAuth,
    workflow_id: str,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List the triggers of a workflow model.

    Returns the raw response: the listing spans every trigger type, not
    only TIMER, so it is not converted into ``DomoWorkflow_Trigger_Timer``.

    Args:
        auth: Authenticated Domo session.
        workflow_id: Workflow UUID.
        return_raw: If True, return the response without raising on failure.
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and
            ``return_raw`` is False.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.list_workflow_triggers_by_model(
        auth=auth,
        workflow_id=workflow_id,
        return_raw=return_raw,
        context=context,
    )

list_trigger_types async classmethod

list_trigger_types(
    auth: DomoAuth,
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List the trigger types Domo offers for workflows.

Returns the raw response: the listing spans every trigger type, not only TIMER, so it is not converted into DomoWorkflow_Trigger_Timer.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
return_raw	bool	If True, return the response without raising on failure.	False
debug_api	bool	If True, enable API debug output.	False
session	AsyncClient | None	Optional shared HTTP client.	None
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides.	{}

Returns:

Type	Description
ResponseGetData	The API response.

Raises:

Type	Description
Workflow_GET_Error	If the response is not successful and return_raw is False.

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

@classmethod
async def list_trigger_types(
    cls,
    auth: DomoAuth,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List the trigger types Domo offers for workflows.

    Returns the raw response: the listing spans every trigger type, not
    only TIMER, so it is not converted into ``DomoWorkflow_Trigger_Timer``.

    Args:
        auth: Authenticated Domo session.
        return_raw: If True, return the response without raising on failure.
        debug_api: If True, enable API debug output.
        session: Optional shared HTTP client.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and
            ``return_raw`` is False.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    return await workflow_routes.list_workflow_trigger_types(
        auth=auth,
        return_raw=return_raw,
        context=context,
    )