core

core

DomoWorkflow dataclass

DomoWorkflow(
    auth: DomoAuth,
    id: str,
    raw: dict = dict(),
    name: str | None = None,
    description: str | None = None,
    owner_id: str | None = None,
    created_by: str | None = None,
    created_dt: datetime | None = None,
    updated_by: str | None = None,
    updated_dt: datetime | None = None,
    last_version_updated_dt: datetime | None = None,
    last_version_updated_by: str | None = None,
    current_version: DomoWorkflow_Version | None = None,
)

Bases: DomoEntity

A Domo workflow model and its versions.

Attributes:

Name	Type	Description
id	str	Workflow model UUID.
auth	DomoAuth	Authenticated Domo session.
raw	dict	Raw API payload the entity was built from.
name	str | None	Workflow name.
description	str | None	Workflow description.
owner_id	str | None	User ID of the owner.
created_by	str | None	User ID who created the model.
created_dt	datetime | None	Creation timestamp.
updated_by	str | None	User ID who last updated the model.
updated_dt	datetime | None	Last-update timestamp.
last_version_updated_dt	datetime | None	When a version was last updated.
last_version_updated_by	str | None	User ID who last updated a version.
versions	DomoWorkflow_Versions	Manager for this model's versions.
current_version	DomoWorkflow_Version | None	Most recently fetched version, if any.

display_url property

display_url: str

Return the Domo UI URL for this workflow.

from_dict classmethod

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

Build a workflow from a /v1/models API payload.

Parameters:

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

Returns:

Type	Description
DomoWorkflow	The populated workflow.

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

@classmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]) -> DomoWorkflow:
    """Build a workflow from a ``/v1/models`` API payload.

    Args:
        auth: Authenticated Domo session.
        obj: Raw workflow model dict.

    Returns:
        The populated workflow.
    """
    return cls(
        auth=auth,
        id=obj.get("id", ""),
        raw=obj,
        name=obj.get("name"),
        description=obj.get("description"),
        owner_id=obj.get("owner"),
        created_by=obj.get("createdBy"),
        created_dt=convert_string_to_datetime(obj.get("createdOn")),
        updated_by=obj.get("updatedBy"),
        updated_dt=convert_string_to_datetime(obj.get("updatedOn")),
        last_version_updated_dt=convert_string_to_datetime(
            obj.get("lastVersionUpdatedOn")
        ),
        last_version_updated_by=obj.get("lastVersionUpdatedBy"),
    )

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow | None

Fetch a workflow by ID (DomoEntity contract).

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
id	str	Workflow model UUID.	required
debug_num_stacks_to_drop	int	Stack frames to drop in debug logging.	2
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 | None	The workflow, or None when the lookup is unsuccessful.

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

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow | None:
    """Fetch a workflow by ID (``DomoEntity`` contract).

    Args:
        auth: Authenticated Domo session.
        id: Workflow model UUID.
        debug_num_stacks_to_drop: Stack frames to drop in debug logging.
        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 workflow, or None when the lookup is unsuccessful.
    """
    return await cls.get_entity_by_id(
        auth=auth,
        entity_id=id,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

get_entity_by_id async classmethod

get_entity_by_id(
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow | None

Fetch a workflow model by ID.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authenticated Domo session.	required
entity_id	str	Workflow model UUID.	required
debug_num_stacks_to_drop	int	Stack frames to drop in debug logging.	2
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 | None	The workflow, or None when the lookup is unsuccessful.

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

@classmethod
async def get_entity_by_id(
    cls,
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow | None:
    """Fetch a workflow model by ID.

    Args:
        auth: Authenticated Domo session.
        entity_id: Workflow model UUID.
        debug_num_stacks_to_drop: Stack frames to drop in debug logging.
        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 workflow, or None when the lookup is unsuccessful.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    res = await workflow_routes.get_workflow_by_id(
        auth=auth,
        workflow_id=entity_id,
        context=context,
    )

    if not res.is_success:
        return None

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

get_execution_metrics async

get_execution_metrics(
    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,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> ResponseGetData

Get execution metrics for this workflow model.

Parameters:

Name	Type	Description	Default
model_version	str | None	Only executions of this model version.	None
status	str | None	Only executions with this status.	None
trigger_types	str | None	Only executions started by these trigger types.	None
started_by	str | None	Only executions started by this user ID.	None
started_after	str | None	Only executions started after this timestamp.	None
finished_before	str | None	Only executions finished before this timestamp.	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/core.py

async def get_execution_metrics(
    self,
    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,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get execution metrics for this workflow model.

    Args:
        model_version: Only executions of this model version.
        status: Only executions with this status.
        trigger_types: Only executions started by these trigger types.
        started_by: Only executions started by this user ID.
        started_after: Only executions started after this timestamp.
        finished_before: Only executions finished before this timestamp.
        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.get_workflow_execution_metrics(
        auth=self.auth,
        model_id=self.id,
        model_version=model_version,
        status=status,
        trigger_types=trigger_types,
        started_by=started_by,
        started_after=started_after,
        finished_before=finished_before,
        return_raw=return_raw,
        context=context,
    )

get_locks async

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

Get the locks held on this workflow model.

Parameters:

Name	Type	Description	Default
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/core.py

async def get_locks(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the locks held on this workflow model.

    Args:
        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.get_workflow_locks(
        auth=self.auth,
        model_id=self.id,
        return_raw=return_raw,
        context=context,
    )

get_permissions async

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

Get the permissions on this workflow model.

Parameters:

Name	Type	Description	Default
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/core.py

async def get_permissions(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> rgd.ResponseGetData:
    """Get the permissions on this workflow model.

    Args:
        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.get_workflow_permissions(
        auth=self.auth,
        model_id=self.id,
        return_raw=return_raw,
        context=context,
    )

get_version async

get_version(
    version: str = "1.0.0",
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow_Version | None

Fetch one specific version through the versions manager.

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

async def get_version(
    self,
    version: str = "1.0.0",
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow_Version | None:
    """Fetch one specific version through the versions manager."""
    ver = await self.versions.get_by_version(
        version=version,
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )
    self.current_version = ver
    return ver

list_triggers async

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

List the triggers attached to this workflow model.

Parameters:

Name	Type	Description	Default
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/core.py

async def list_triggers(
    self,
    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 attached to this workflow model.

    Args:
        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=self.auth,
        workflow_id=self.id,
        return_raw=return_raw,
        context=context,
    )

upsert async classmethod

upsert(
    auth: DomoAuth,
    name: str,
    saved_json: str,
    version: str = "1.0.0",
    description: str = "",
    model_id: str | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> DomoWorkflow

Create a model (or reuse an existing one) then PUT the version.

If model_id is not supplied a new model is created via POST.

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

@classmethod
async def upsert(
    cls,
    auth: DomoAuth,
    name: str,
    saved_json: str,
    version: str = "1.0.0",
    description: str = "",
    model_id: str | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> DomoWorkflow:
    """Create a model (or reuse an existing one) then PUT the version.

    If `model_id` is not supplied a new model is created via POST.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    is_new_model = model_id is None
    if is_new_model:
        res = await workflow_routes.upsert_workflow_model(
            auth=auth,
            name=name,
            description=description,
            context=context,
        )
        res.response = res.response or {}
        model_id = res.response.get("id")

    # New models have no versions yet — POST to create; existing models use PUT
    if is_new_model:
        await workflow_routes.create_workflow_version_from_saved_json(
            auth=auth,
            model_id=model_id,
            version=version,
            saved_json=saved_json,
            context=context,
        )
    else:
        await cls._assert_version_is_updatable(
            auth=auth,
            model_id=model_id,
            version=version,
            context=context,
        )
        await workflow_routes.upsert_workflow_version(
            auth=auth,
            model_id=model_id,
            version=version,
            saved_json=saved_json,
            context=context,
        )

    return await cls.get_entity_by_id(
        auth=auth,
        entity_id=model_id,
        context=context,
    )

validate_model_version async classmethod

validate_model_version(
    auth: DomoAuth,
    model_id: str,
    version: str,
    use_domo_validator: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> tuple[bool, list[str]]

Fetch a workflow version by ID and validate its graph topology.

Runs local structural validation (connection rules + True/False branch labels on every conditional gateway) by default. Pass use_domo_validator=True to additionally call Domo's official /api/workflow/v2/models/validate endpoint and fold in any errors it reports.

Returns (is_valid, errors) — errors is empty when is_valid is True.

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

@classmethod
async def validate_model_version(
    cls,
    auth: DomoAuth,
    model_id: str,
    version: str,
    use_domo_validator: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> tuple[bool, list[str]]:
    """Fetch a workflow version by ID and validate its graph topology.

    Runs local structural validation (connection rules + True/False branch
    labels on every conditional gateway) by default. Pass
    ``use_domo_validator=True`` to additionally call Domo's official
    ``/api/workflow/v2/models/validate`` endpoint and fold in any errors
    it reports.

    Returns (is_valid, errors) — errors is empty when is_valid is True.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    workflow = await cls.get_entity_by_id(
        auth=auth, entity_id=model_id, context=context
    )
    if workflow is None:
        return False, [f"workflow not found: {model_id}"]

    ver = await workflow.get_version(version=version, context=context)
    if ver is None or not ver.tiles:
        return False, [f"workflow version not found: {model_id}@{version}"]

    is_valid, errors = ver.validate_definition()
    errors = list(errors)

    if use_domo_validator:
        try:
            res = await ver.validate_with_domo(auth=auth, context=context)
            if not res.is_success:
                is_valid = False
                errors.append(
                    f"Domo validator rejected the workflow: {res.response}"
                )
        except Exception as exc:  # noqa: BLE001
            is_valid = False
            errors.append(f"Domo validator call failed: {exc}")

    return is_valid, errors

DomoWorkflow_DeployedVersionError

DomoWorkflow_DeployedVersionError(
    model_id: str,
    version: str,
    deployed_on: datetime | None = None,
    active: bool | None = None,
    cls_instance: Any | None = None,
)

Bases: ClassError

Raised when attempting to update a deployed workflow version in place.

Build the error message for a deployed, non-updatable version.

Parameters:

Name	Type	Description	Default
model_id	str	Workflow model UUID.	required
version	str	Semver string of the deployed version.	required
deployed_on	datetime | None	When the version was deployed, if known.	None
active	bool | None	Whether the version is currently active.	None
cls_instance	Any | None	Optional entity that raised the error, for context.	None

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

def __init__(
    self,
    model_id: str,
    version: str,
    deployed_on: dt.datetime | None = None,
    active: bool | None = None,
    cls_instance: Any | None = None,
) -> None:
    """Build the error message for a deployed, non-updatable version.

    Args:
        model_id: Workflow model UUID.
        version: Semver string of the deployed version.
        deployed_on: When the version was deployed, if known.
        active: Whether the version is currently active.
        cls_instance: Optional entity that raised the error, for context.
    """
    deployed_txt = deployed_on.isoformat() if deployed_on else "unknown"
    active_txt = "true" if active else "false"
    super().__init__(
        cls_instance=cls_instance,
        entity_id=model_id,
        message=(
            f"Workflow version '{version}' is deployed (active={active_txt}, "
            f"deployedOn={deployed_txt}) and cannot be updated in place. "
            "Create a new version (for example, 1.0.1) and publish that version instead."
        ),
    )

DomoWorkflows dataclass

DomoWorkflows(
    auth: DomoAuth,
    workflows: list[DomoWorkflow] | None = None,
)

Manager for listing and searching workflow models.

Attributes:

Name	Type	Description
auth	DomoAuth	Authenticated Domo session.
workflows	list[DomoWorkflow] | None	Cached list from the last get call, if any.

get async

get(
    return_raw: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> list[DomoWorkflow] | ResponseGetData

List all workflow models and cache them on self.workflows.

Parameters:

Name	Type	Description	Default
return_raw	bool	If True, return the raw route response instead of entities.	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
list[DomoWorkflow] | ResponseGetData	The workflows, or the raw response when return_raw is True.

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

async def get(
    self,
    return_raw: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> list[DomoWorkflow] | rgd.ResponseGetData:
    """List all workflow models and cache them on ``self.workflows``.

    Args:
        return_raw: If True, return the raw route response instead of entities.
        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 workflows, or the raw response when ``return_raw`` is True.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    res = await workflow_routes.list_workflows(
        auth=self.auth,
        context=context,
    )

    if return_raw:
        return res

    models_raw = res.response if res.is_success else []
    self.workflows = [
        DomoWorkflow.from_dict(auth=self.auth, obj=obj)
        for obj in (models_raw or [])
    ]
    return self.workflows

get_models_usage async

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

List workflow usage of an entity (for example a Code Engine package).

Parameters:

Name	Type	Description	Default
entity_type	str | None	Entity type to look up usage for.	None
entity_id	str | None	Entity ID to look up usage for.	None
parts	str | None	Optional response parts selector.	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/core.py

async def get_models_usage(
    self,
    entity_type: str | None = None,
    entity_id: str | None = None,
    parts: 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 workflow usage of an entity (for example a Code Engine package).

    Args:
        entity_type: Entity type to look up usage for.
        entity_id: Entity ID to look up usage for.
        parts: Optional response parts selector.
        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.get_workflow_models_usage(
        auth=self.auth,
        entity_type=entity_type,
        entity_id=entity_id,
        parts=parts,
        return_raw=return_raw,
        context=context,
    )

search_by_name

search_by_name(name: str) -> list[DomoWorkflow]

Filter the cached workflow list by name (case-insensitive substring).

Parameters:

Name	Type	Description	Default
name	str	Substring to match against workflow names.	required

Returns:

Type	Description
list[DomoWorkflow]	Matching workflows.

Raises:

Type	Description
RuntimeError	If get has not populated the list yet.

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

def search_by_name(self, name: str) -> list[DomoWorkflow]:
    """Filter the cached workflow list by name (case-insensitive substring).

    Args:
        name: Substring to match against workflow names.

    Returns:
        Matching workflows.

    Raises:
        RuntimeError: If ``get`` has not populated the list yet.
    """
    if self.workflows is None:
        raise RuntimeError("Call get() first to populate the workflow list.")
    name_lower = name.lower()
    return [w for w in self.workflows if name_lower in (w.name or "").lower()]