workflows

workflows

Workflow_CRUD_Error

Workflow_CRUD_Error(
    operation: str,
    workflow_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when workflow create, update, delete, or trigger operations fail.

Source code in src/crew_dcs/routes/workflows.py

def __init__(
    self,
    operation: str,
    workflow_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or f"Workflow {operation} operation failed",
        entity_id=workflow_id,
        res=res,
        **kwargs,
    )

Workflow_GET_Error

Workflow_GET_Error(
    workflow_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs
)

Bases: RouteError

Raised when workflow retrieval operations fail.

Source code in src/crew_dcs/routes/workflows.py

def __init__(
    self,
    workflow_id: str | None = None,
    message: str | None = None,
    res=None,
    **kwargs,
):
    super().__init__(
        message=message or "Workflow retrieval failed",
        entity_id=workflow_id,
        res=res,
        **kwargs,
    )

create_workflow async

create_workflow(
    auth: DomoAuth,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new workflow.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_workflow(
    auth: DomoAuth,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new workflow."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/models"
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body=body or {},
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(operation="create", res=res)

    return res

create_workflow_trigger async

create_workflow_trigger(
    auth: DomoAuth,
    name: str,
    trigger_type: str,
    model_id: str,
    model_version: str,
    entity_type: str,
    entity_id: str,
    metadata: dict,
    default_values: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

POST creates a trigger (e.g. TIMER) that fires a workflow model/version.

entity_type/entity_id scope trigger ownership — for TIMER triggers Domo requires entity_type="USER" and entity_id=the creating user's ID (a token can only create timer triggers for itself). metadata must include a "type" discriminator matching trigger_type (e.g. {"type": "TIMER", "cron": "0 0 2 * * ?", "timeZone": "UTC"}).

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_workflow_trigger(
    auth: DomoAuth,
    name: str,
    trigger_type: str,
    model_id: str,
    model_version: str,
    entity_type: str,
    entity_id: str,
    metadata: dict,
    default_values: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """POST creates a trigger (e.g. TIMER) that fires a workflow model/version.

    `entity_type`/`entity_id` scope trigger ownership — for TIMER triggers
    Domo requires entity_type="USER" and entity_id=the creating user's ID
    (a token can only create timer triggers for itself).
    `metadata` must include a "type" discriminator matching `trigger_type`
    (e.g. {"type": "TIMER", "cron": "0 0 2 * * ?", "timeZone": "UTC"}).
    """
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/triggers"
    body = {
        "name": name,
        "type": trigger_type,
        "modelId": model_id,
        "modelVersion": model_version,
        "entityType": entity_type,
        "entityId": entity_id,
        "defaultValues": default_values or {},
        "metadata": metadata,
    }
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(
            operation="create_trigger", workflow_id=model_id, res=res
        )

    return res

create_workflow_version async

create_workflow_version(
    auth: DomoAuth,
    workflow_id: str,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a new version of a workflow.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_workflow_version(
    auth: DomoAuth,
    workflow_id: str,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a new version of a workflow."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/models/{workflow_id}/versions"
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body=body or {},
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(
            operation="create_version", workflow_id=workflow_id, res=res
        )

    return res

create_workflow_version_from_saved_json async

create_workflow_version_from_saved_json(
    auth: DomoAuth,
    model_id: str,
    version: str,
    saved_json: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

POST creates a brand-new version on a model that has no versions yet.

Uses the v1 API with a savedJson payload. For the generic v2 body-dict route see create_workflow_version.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def create_workflow_version_from_saved_json(
    auth: DomoAuth,
    model_id: str,
    version: str,
    saved_json: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """POST creates a brand-new version on a model that has no versions yet.

    Uses the v1 API with a `savedJson` payload. For the generic v2 body-dict
    route see `create_workflow_version`.
    """
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models/{model_id}/versions"
    body = {"savedJson": saved_json, "version": version}
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(
            operation="create_version", workflow_id=model_id, res=res
        )

    return res

delete_workflow async

delete_workflow(
    auth: DomoAuth,
    workflow_id: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Delete a workflow.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_workflow(
    auth: DomoAuth,
    workflow_id: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Delete a workflow."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models/{workflow_id}"
    res = await gd.get_data(
        auth=auth,
        method="DELETE",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(operation="delete", workflow_id=workflow_id, res=res)

    return res

delete_workflow_trigger async

delete_workflow_trigger(
    auth: DomoAuth,
    trigger_id: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

DELETE soft-deletes a trigger (sets deletedAt; stops future firings).

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def delete_workflow_trigger(
    auth: DomoAuth,
    trigger_id: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """DELETE soft-deletes a trigger (sets deletedAt; stops future firings)."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/triggers/{trigger_id}"

    res = await gd.get_data(
        auth=auth,
        method="DELETE",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(
            operation="delete_trigger", workflow_id=trigger_id, res=res
        )

    return res

get_workflow_by_id async

get_workflow_by_id(
    auth: DomoAuth,
    workflow_id: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get a workflow by its ID.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_by_id(
    auth: DomoAuth,
    workflow_id: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a workflow by its ID."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models/{workflow_id}"
    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_GET_Error(workflow_id=workflow_id, res=res)

    return res

get_workflow_execution_by_id async

get_workflow_execution_by_id(
    auth: DomoAuth,
    execution_id: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get a workflow execution by its ID.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_execution_by_id(
    auth: DomoAuth,
    execution_id: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a workflow execution by its ID."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/executions/{execution_id}"
    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_GET_Error(workflow_id=execution_id, res=res)

    return res

get_workflow_execution_metrics async

get_workflow_execution_metrics(
    auth: DomoAuth,
    model_id: str,
    model_version: str | None = None,
    status: str | None = None,
    trigger_types: str | None = None,
    started_by: str | None = None,
    started_after: str | None = None,
    finished_before: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get execution metrics for a workflow (Postman: Get Execution Metrics).

GET /api/workflow/v2/executions/metrics. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
model_id	str	Workflow UUID (sent as modelId; required by Domo).	required
model_version	str | None	Optional semantic version (modelVersion).	None
status	str | None	Optional comma-separated statuses (COMPLETED, IN_PROGRESS, CANCELLED, FAILED).	None
trigger_types	str | None	Optional comma-separated trigger types (triggerTypes).	None
started_by	str | None	Optional user ID (startedBy).	None
started_after	str | None	Optional ISO 8601 timestamp (startedAfter).	None
finished_before	str | None	Optional ISO 8601 timestamp (finishedBefore).	None
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_execution_metrics(
    auth: DomoAuth,
    model_id: str,
    model_version: str | None = None,
    status: str | None = None,
    trigger_types: str | None = None,
    started_by: str | None = None,
    started_after: str | None = None,
    finished_before: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get execution metrics for a workflow (Postman: Get Execution Metrics).

    ``GET /api/workflow/v2/executions/metrics``. Response shape is unverified
    against a live Domo.

    Args:
        auth: Authenticated Domo session.
        model_id: Workflow UUID (sent as ``modelId``; required by Domo).
        model_version: Optional semantic version (``modelVersion``).
        status: Optional comma-separated statuses (COMPLETED, IN_PROGRESS,
            CANCELLED, FAILED).
        trigger_types: Optional comma-separated trigger types (``triggerTypes``).
        started_by: Optional user ID (``startedBy``).
        started_after: Optional ISO 8601 timestamp (``startedAfter``).
        finished_before: Optional ISO 8601 timestamp (``finishedBefore``).
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    params = _drop_none(
        {
            "modelId": model_id,
            "modelVersion": model_version,
            "status": status,
            "triggerTypes": trigger_types,
            "startedBy": started_by,
            "startedAfter": started_after,
            "finishedBefore": finished_before,
        }
    )
    return await _get_workflow_path(
        auth,
        "/v2/executions/metrics",
        entity_id=model_id,
        params=params,
        return_raw=return_raw,
        context=context,
    )

get_workflow_execution_outputs async

get_workflow_execution_outputs(
    auth: DomoAuth,
    execution_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the outputs of a workflow execution (Postman: Get Execution Outputs).

GET /api/workflow/v2/executions/{execution_id}/outputs. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
execution_id	str	Workflow execution UUID.	required
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_execution_outputs(
    auth: DomoAuth,
    execution_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the outputs of a workflow execution (Postman: Get Execution Outputs).

    ``GET /api/workflow/v2/executions/{execution_id}/outputs``. Response shape
    is unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        execution_id: Workflow execution UUID.
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    return await _get_workflow_path(
        auth,
        f"/v2/executions/{execution_id}/outputs",
        entity_id=execution_id,
        params=None,
        return_raw=return_raw,
        context=context,
    )

get_workflow_locks async

get_workflow_locks(
    auth: DomoAuth,
    model_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the locks held on a workflow model (Postman: Get Locks).

GET /api/workflow/v1/models/{model_id}/locks. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
model_id	str	Workflow model UUID.	required
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_locks(
    auth: DomoAuth,
    model_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the locks held on a workflow model (Postman: Get Locks).

    ``GET /api/workflow/v1/models/{model_id}/locks``. Response shape is
    unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        model_id: Workflow model UUID.
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    return await _get_workflow_path(
        auth,
        f"/v1/models/{model_id}/locks",
        entity_id=model_id,
        params=None,
        return_raw=return_raw,
        context=context,
    )

get_workflow_models_usage async

get_workflow_models_usage(
    auth: DomoAuth,
    entity_type: str | None = None,
    entity_id: str | None = None,
    parts: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

List workflow usage of an entity (Postman: List Code Engine Package Usage).

GET /api/workflow/v2/models/usage. The Postman example uses entityType=CODEENGINE_FUNCTION, a Code Engine package UUID as entityId and parts=items. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
entity_type	str | None	Optional entity type (entityType).	None
entity_id	str | None	Optional entity UUID (entityId).	None
parts	str | None	Optional response parts (for example items).	None
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_models_usage(
    auth: DomoAuth,
    entity_type: str | None = None,
    entity_id: str | None = None,
    parts: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List workflow usage of an entity (Postman: List Code Engine Package Usage).

    ``GET /api/workflow/v2/models/usage``. The Postman example uses
    ``entityType=CODEENGINE_FUNCTION``, a Code Engine package UUID as
    ``entityId`` and ``parts=items``. Response shape is unverified against a
    live Domo.

    Args:
        auth: Authenticated Domo session.
        entity_type: Optional entity type (``entityType``).
        entity_id: Optional entity UUID (``entityId``).
        parts: Optional response parts (for example ``items``).
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    params = _drop_none(
        {"entityType": entity_type, "entityId": entity_id, "parts": parts}
    )
    return await _get_workflow_path(
        auth,
        "/v2/models/usage",
        entity_id=entity_id,
        params=params,
        return_raw=return_raw,
        context=context,
    )

get_workflow_permissions async

get_workflow_permissions(
    auth: DomoAuth,
    model_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the permissions on a workflow model (Postman: Get Permissions).

GET /api/workflow/v1/models/{model_id}/permissions. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
model_id	str	Workflow model UUID.	required
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_permissions(
    auth: DomoAuth,
    model_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the permissions on a workflow model (Postman: Get Permissions).

    ``GET /api/workflow/v1/models/{model_id}/permissions``. Response shape is
    unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        model_id: Workflow model UUID.
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    return await _get_workflow_path(
        auth,
        f"/v1/models/{model_id}/permissions",
        entity_id=model_id,
        params=None,
        return_raw=return_raw,
        context=context,
    )

get_workflow_trigger_form async

get_workflow_trigger_form(
    auth: DomoAuth,
    model_id: str,
    version: str,
    trigger_id: str | None = None,
    deployment_id: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the trigger form of a workflow version (Postman: Get Trigger Form).

GET /api/workflow/v2/models/{model_id}/versions/{version}/form. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
model_id	str	Workflow model UUID.	required
version	str	Workflow version number.	required
trigger_id	str | None	Optional trigger UUID (triggerId).	None
deployment_id	str | None	Optional deployment UUID (deploymentId).	None
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_trigger_form(
    auth: DomoAuth,
    model_id: str,
    version: str,
    trigger_id: str | None = None,
    deployment_id: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the trigger form of a workflow version (Postman: Get Trigger Form).

    ``GET /api/workflow/v2/models/{model_id}/versions/{version}/form``.
    Response shape is unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        model_id: Workflow model UUID.
        version: Workflow version number.
        trigger_id: Optional trigger UUID (``triggerId``).
        deployment_id: Optional deployment UUID (``deploymentId``).
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    params = _drop_none({"triggerId": trigger_id, "deploymentId": deployment_id})
    return await _get_workflow_path(
        auth,
        f"/v2/models/{model_id}/versions/{version}/form",
        entity_id=model_id,
        params=params,
        return_raw=return_raw,
        context=context,
    )

get_workflow_version async

get_workflow_version(
    auth: DomoAuth,
    workflow_id: str,
    version_number: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Get a specific version of a workflow.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_version(
    auth: DomoAuth,
    workflow_id: str,
    version_number: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Get a specific version of a workflow."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/models/{workflow_id}/versions/{version_number}"
    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_GET_Error(workflow_id=workflow_id, res=res)

    return res

get_workflow_version_definition async

get_workflow_version_definition(
    auth: DomoAuth,
    workflow_id: str,
    version_number: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get the definition of a workflow version (Postman: Get Version Definition).

GET /api/workflow/v2/models/{workflow_id}/versions/{version_number}/definition. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
workflow_id	str	Workflow UUID.	required
version_number	str	Workflow version number.	required
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_workflow_version_definition(
    auth: DomoAuth,
    workflow_id: str,
    version_number: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the definition of a workflow version (Postman: Get Version Definition).

    ``GET /api/workflow/v2/models/{workflow_id}/versions/{version_number}/definition``.
    Response shape is unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        workflow_id: Workflow UUID.
        version_number: Workflow version number.
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    return await _get_workflow_path(
        auth,
        f"/v2/models/{workflow_id}/versions/{version_number}/definition",
        entity_id=workflow_id,
        params=None,
        return_raw=return_raw,
        context=context,
    )

list_workflow_executions async

list_workflow_executions(
    auth: DomoAuth,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

List all workflow executions.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_workflow_executions(
    auth: DomoAuth,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """List all workflow executions."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/executions"
    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_GET_Error(res=res)

    return res

list_workflow_trigger_types async

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

List the available workflow trigger types (Postman: List Trigger Types).

GET /api/workflow/v2/triggers/types. Response shape is unverified against a live Domo.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_workflow_trigger_types(
    auth: DomoAuth,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List the available workflow trigger types (Postman: List Trigger Types).

    ``GET /api/workflow/v2/triggers/types``. Response shape is unverified
    against a live Domo.

    Args:
        auth: Authenticated Domo session.
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    return await _get_workflow_path(
        auth,
        "/v2/triggers/types",
        entity_id=None,
        params=None,
        return_raw=return_raw,
        context=context,
    )

list_workflow_triggers_by_entity async

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

List triggers attached to an entity (Postman: List Triggers by Entity).

GET /api/workflow/v2/triggers/entity/{entity_type}/{entity_id}. Response shape is unverified against a live Domo.

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 (entityVersions).	None
return_raw	bool	If True, return the response without raising on failure.	False
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_workflow_triggers_by_entity(
    auth: DomoAuth,
    entity_type: str,
    entity_id: str,
    entity_versions: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List triggers attached to an entity (Postman: List Triggers by Entity).

    ``GET /api/workflow/v2/triggers/entity/{entity_type}/{entity_id}``.
    Response shape is unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        entity_type: Entity type path segment.
        entity_id: Entity UUID.
        entity_versions: Optional entity versions to filter by (``entityVersions``).
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    params = _drop_none({"entityVersions": entity_versions})
    return await _get_workflow_path(
        auth,
        f"/v2/triggers/entity/{entity_type}/{entity_id}",
        entity_id=entity_id,
        params=params,
        return_raw=return_raw,
        context=context,
    )

list_workflow_triggers_by_model async

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

List the triggers of a workflow (Postman: List Triggers).

GET /api/workflow/v2/triggers/model/{workflow_id}. Response shape is unverified against a live Domo.

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
context	RouteContext | None	Optional pre-built route context.	None
**context_kwargs	Any	Extra context overrides consumed by route_function.	{}

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/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_workflow_triggers_by_model(
    auth: DomoAuth,
    workflow_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """List the triggers of a workflow (Postman: List Triggers).

    ``GET /api/workflow/v2/triggers/model/{workflow_id}``. Response shape is
    unverified against a live Domo.

    Args:
        auth: Authenticated Domo session.
        workflow_id: Workflow UUID.
        return_raw: If True, return the response without raising on failure.
        context: Optional pre-built route context.
        **context_kwargs: Extra context overrides consumed by ``route_function``.

    Returns:
        The API response.

    Raises:
        Workflow_GET_Error: If the response is not successful and ``return_raw`` is False.
    """
    return await _get_workflow_path(
        auth,
        f"/v2/triggers/model/{workflow_id}",
        entity_id=workflow_id,
        params=None,
        return_raw=return_raw,
        context=context,
    )

list_workflow_versions async

list_workflow_versions(
    auth: DomoAuth,
    model_id: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

List all versions of a workflow model.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_workflow_versions(
    auth: DomoAuth,
    model_id: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """List all versions of a workflow model."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models/{model_id}/versions"
    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_GET_Error(workflow_id=model_id, res=res)

    return res

list_workflows async

list_workflows(
    auth: DomoAuth,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

List all workflows.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def list_workflows(
    auth: DomoAuth,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """List all workflows."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models"
    res = await gd.get_data(
        auth=auth,
        method="GET",
        url=url,
        context=context,
    )

    if not res.is_success:
        raise Workflow_GET_Error(res=res)

    return res

release_workflow_version async

release_workflow_version(
    auth: DomoAuth,
    model_id: str,
    version: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

POST releases (deploys/activates) a workflow version.

The version must have a valid savedJson with proper style/zIndex fields for the release to succeed. After release, the version becomes active and can receive triggers (TIMER, MANUAL, etc.).

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def release_workflow_version(
    auth: DomoAuth,
    model_id: str,
    version: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """POST releases (deploys/activates) a workflow version.

    The version must have a valid savedJson with proper style/zIndex fields
    for the release to succeed. After release, the version becomes active
    and can receive triggers (TIMER, MANUAL, etc.).
    """
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/models/{model_id}/versions/{version}/release"
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body={},
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(operation="release", workflow_id=model_id, res=res)

    return res

update_workflow async

update_workflow(
    auth: DomoAuth,
    workflow_id: str,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Update a workflow.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def update_workflow(
    auth: DomoAuth,
    workflow_id: str,
    body: dict | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Update a workflow."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models/{workflow_id}"
    res = await gd.get_data(
        auth=auth,
        method="PUT",
        url=url,
        body=body or {},
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(operation="update", workflow_id=workflow_id, res=res)

    return res

upsert_workflow_model async

upsert_workflow_model(
    auth: DomoAuth,
    name: str,
    description: str = "",
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Create a workflow model shell (no versions) via the v1 API.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upsert_workflow_model(
    auth: DomoAuth,
    name: str,
    description: str = "",
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Create a workflow model shell (no versions) via the v1 API."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models"
    body = {"name": name, "description": description}
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(operation="upsert_model", res=res)

    return res

upsert_workflow_version async

upsert_workflow_version(
    auth: DomoAuth,
    model_id: str,
    version: str,
    saved_json: str,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

PUT updates an existing version with the provided savedJson string.

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def upsert_workflow_version(
    auth: DomoAuth,
    model_id: str,
    version: str,
    saved_json: str,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """PUT updates an existing version with the provided savedJson string."""
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v1/models/{model_id}/versions/{version}"
    body = {"savedJson": saved_json}
    res = await gd.get_data(
        auth=auth,
        method="PUT",
        url=url,
        body=body,
        context=context,
    )

    if not res.is_success:
        message = None
        if res.status == 403:
            message = (
                "Unable to update workflow version. Validate that the version is not "
                "deployed/active and that your token has edit permission. "
                "If the version is deployed, create a new version instead of updating "
                "the deployed version in place."
            )
            try:
                version_res = await get_workflow(
                    auth=auth,
                    model_id=model_id,
                    version_id=version,
                    context=context,
                )
                version_meta = version_res.response or {}
                message += (
                    " "
                    f"Current metadata: active={version_meta.get('active')}, "
                    f"deployedOn={version_meta.get('deployedOn')}."
                )
            except Exception:  # noqa: BLE001
                # Keep the core 403 hint even if metadata lookup fails.
                pass
        raise Workflow_CRUD_Error(
            operation="upsert_version", workflow_id=model_id, res=res, message=message
        )

    return res

validate_workflow_model async

validate_workflow_model(
    auth: DomoAuth,
    body: dict,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Validate a workflow graph payload using Domo's official validator API.

Endpoint: POST /api/workflow/v2/models/validate

Source code in src/crew_dcs/routes/workflows.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def validate_workflow_model(
    auth: DomoAuth,
    body: dict,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Validate a workflow graph payload using Domo's official validator API.

    Endpoint: ``POST /api/workflow/v2/models/validate``
    """
    url = f"https://{auth.domo_instance}.domo.com/api/workflow/v2/models/validate"
    res = await gd.get_data(
        auth=auth,
        method="POST",
        url=url,
        body=body,
        context=context,
    )

    if not res.is_success:
        raise Workflow_CRUD_Error(operation="validate", res=res)

    return res