DomoWorkflow

DomoWorkflow ¶

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_Definition `dataclass` ¶

DomoWorkflow_Definition(
    schema_version: int = 2,
    inputs: dict[str, DomoWorkflow_Parameter] = dict(),
    outputs: dict[str, DomoWorkflow_Parameter] = dict(),
    tiles: list[DomoWorkflow_Tile] = list(),
    edges: list[DomoWorkflow_Edge] = list(),
    data_list: list = list(),
    Design: DomoWorkflow_Design | None = None,
    raw: dict = dict(),
)

Parsed representation of a workflow's savedJson.

Separates designElements into typed tiles and edges, and exposes helpers for programmatic construction.

layouts `property` `writable` ¶

layouts: list

Alias for data_list to align with Design/Tiles/Layouts terminology.

add_boolean_branch_edges ¶

add_boolean_branch_edges(
    gateway_id: str,
    true_target_id: str,
    false_target_id: str,
    *,
    true_label: str = "True",
    false_label: str = "False",
    condition_type: str = "Basic"
) -> tuple[DomoWorkflow_Edge, DomoWorkflow_Edge]

Add the required True/False branches for a conditional gateway.

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

def add_boolean_branch_edges(
    self,
    gateway_id: str,
    true_target_id: str,
    false_target_id: str,
    *,
    true_label: str = "True",
    false_label: str = "False",
    condition_type: str = "Basic",
) -> tuple[DomoWorkflow_Edge, DomoWorkflow_Edge]:
    """Add the required True/False branches for a conditional gateway."""
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    true_edge, false_edge = self.Design.add_boolean_branch_edges(
        gateway_id=gateway_id,
        true_target_id=true_target_id,
        false_target_id=false_target_id,
        true_label=true_label,
        false_label=false_label,
        condition_type=condition_type,
    )
    self._sync_legacy_fields_from_design()
    return true_edge, false_edge

add_edge ¶

add_edge(
    source_id: str,
    target_id: str,
    title: str | None = None,
    condition_type: str | None = None,
) -> DomoWorkflow_Edge

Create and append an edge between two tile IDs.

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

def add_edge(
    self,
    source_id: str,
    target_id: str,
    title: str | None = None,
    condition_type: str | None = None,
) -> DomoWorkflow_Edge:
    """Create and append an edge between two tile IDs."""
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    edge = self.Design.add_edge(
        source_id=source_id,
        target_id=target_id,
        title=title,
        condition_type=condition_type,
    )
    self._sync_legacy_fields_from_design()
    return edge

add_input ¶

add_input(param: DomoWorkflow_Parameter) -> None

Add a parameter to the schema inputs.

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

def add_input(self, param: DomoWorkflow_Parameter) -> None:
    """Add a parameter to the schema inputs."""
    self.inputs[param.param_name] = param

add_output ¶

add_output(param: DomoWorkflow_Parameter) -> None

Add a parameter to the schema outputs.

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

def add_output(self, param: DomoWorkflow_Parameter) -> None:
    """Add a parameter to the schema outputs."""
    self.outputs[param.param_name] = param

add_tile ¶

add_tile(
    tile: DomoWorkflow_Tile,
    x: float | None = None,
    y: float | None = None,
) -> DomoWorkflow_Tile

Append a tile, auto-positioning if x/y not given (stack vertically).

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

def add_tile(
    self,
    tile: DomoWorkflow_Tile,
    x: float | None = None,
    y: float | None = None,
) -> DomoWorkflow_Tile:
    """Append a tile, auto-positioning if x/y not given (stack vertically)."""
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    tile = self.Design.add_tile(tile=tile, x=x, y=y)
    self._sync_legacy_fields_from_design()
    return tile

empty `classmethod` ¶

empty() -> DomoWorkflow_Definition

Create an empty definition suitable for a brand-new workflow.

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

@classmethod
def empty(cls) -> DomoWorkflow_Definition:
    """Create an empty definition suitable for a brand-new workflow."""
    return cls()

from_saved_json `classmethod` ¶

from_saved_json(
    saved_json_str: str,
) -> DomoWorkflow_Definition

Parse the savedJson string stored on a version object.

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

@classmethod
def from_saved_json(cls, saved_json_str: str) -> DomoWorkflow_Definition:
    """Parse the savedJson string stored on a version object."""
    raw = json.loads(saved_json_str)
    return cls._from_raw(raw)

get_connection_validation_errors ¶

get_connection_validation_errors(
    *,
    logic_tile_types: tuple[str, ...] = (
        "conditionalGatewayNode",
    ),
    allow_dangling_tile_types: tuple[str, ...] = (
        "endNode",
    ),
    required_logic_outputs: int = 2
) -> list[str]

Return graph-connection validation errors without raising.

Rules: - Only tiles in allow_dangling_tile_types may have zero outgoing edges. - Tiles in logic_tile_types must have exactly required_logic_outputs outgoing edges. - Edges must reference existing source/target tile IDs.

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

def get_connection_validation_errors(
    self,
    *,
    logic_tile_types: tuple[str, ...] = ("conditionalGatewayNode",),
    allow_dangling_tile_types: tuple[str, ...] = ("endNode",),
    required_logic_outputs: int = 2,
) -> list[str]:
    """Return graph-connection validation errors without raising.

    Rules:
    - Only tiles in ``allow_dangling_tile_types`` may have zero outgoing edges.
    - Tiles in ``logic_tile_types`` must have exactly ``required_logic_outputs``
      outgoing edges.
    - Edges must reference existing source/target tile IDs.
    """
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    return self.Design.get_connection_validation_errors(
        logic_tile_types=logic_tile_types,
        allow_dangling_tile_types=allow_dangling_tile_types,
        required_logic_outputs=required_logic_outputs,
    )

set_guarded_codeengine_flow ¶

set_guarded_codeengine_flow(
    *,
    package_id: str,
    package_version: str,
    success_functions: list[str],
    failure_notification_emails: list[str] | None = None,
    inputs_by_function: (
        dict[str, list[DomoWorkflow_Parameter]] | None
    ) = None,
    start_title: str = "Start",
    end_title: str = "End",
    notification_title: str = "Notify Owners",
    notification_message: str = "Workflow preflight failed",
    start_x: float = 80.0,
    start_y: float = 80.0,
    y_step: float = 150.0
) -> None

Build a guarded workflow graph via Design manager.

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

def set_guarded_codeengine_flow(
    self,
    *,
    package_id: str,
    package_version: str,
    success_functions: list[str],
    failure_notification_emails: list[str] | None = None,
    inputs_by_function: dict[str, list[DomoWorkflow_Parameter]] | None = None,
    start_title: str = "Start",
    end_title: str = "End",
    notification_title: str = "Notify Owners",
    notification_message: str = "Workflow preflight failed",
    start_x: float = 80.0,
    start_y: float = 80.0,
    y_step: float = 150.0,
) -> None:
    """Build a guarded workflow graph via Design manager."""
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    self.Design.set_guarded_codeengine_flow(
        package_id=package_id,
        package_version=package_version,
        success_functions=success_functions,
        failure_notification_emails=failure_notification_emails,
        inputs_by_function=inputs_by_function,
        start_title=start_title,
        end_title=end_title,
        notification_title=notification_title,
        notification_message=notification_message,
        start_x=start_x,
        start_y=start_y,
        y_step=y_step,
    )
    self._sync_legacy_fields_from_design()

set_linear_codeengine_flow ¶

set_linear_codeengine_flow(
    *,
    package_id: str,
    package_version: str,
    function_names: list[str],
    inputs_by_function: (
        dict[str, list[DomoWorkflow_Parameter]] | None
    ) = None,
    start_title: str = "Start",
    end_title: str = "End",
    start_x: float = 80.0,
    start_y: float = 80.0,
    y_step: float = 150.0
) -> None

Replace current graph with a linear CodeEngine flow via Design manager.

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

def set_linear_codeengine_flow(
    self,
    *,
    package_id: str,
    package_version: str,
    function_names: list[str],
    inputs_by_function: dict[str, list[DomoWorkflow_Parameter]] | None = None,
    start_title: str = "Start",
    end_title: str = "End",
    start_x: float = 80.0,
    start_y: float = 80.0,
    y_step: float = 150.0,
) -> None:
    """Replace current graph with a linear CodeEngine flow via Design manager."""
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    self.Design.set_linear_codeengine_flow(
        package_id=package_id,
        package_version=package_version,
        function_names=function_names,
        inputs_by_function=inputs_by_function,
        start_title=start_title,
        end_title=end_title,
        start_x=start_x,
        start_y=start_y,
        y_step=y_step,
    )
    self._sync_legacy_fields_from_design()

to_saved_json ¶

to_saved_json() -> str

Serialize back to the savedJson string expected by the API.

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

def to_saved_json(self) -> str:
    """Serialize back to the savedJson string expected by the API."""
    self._sync_design_from_legacy_fields()
    design_payload = (
        self.Design.to_dict()
        if self.Design is not None
        else {
            "version": self.schema_version,
            "dataList": self.data_list,
            "designElements": [t.to_dict() for t in self.tiles]
            + [e.to_dict() for e in self.edges],
        }
    )
    payload = {
        "schema": {
            "inputs": {k: v.to_dict() for k, v in self.inputs.items()},
            "outputs": {k: v.to_dict() for k, v in self.outputs.items()},
        },
        "version": design_payload["version"],
        "dataList": design_payload["dataList"],
        "designElements": design_payload["designElements"],
    }
    return json.dumps(payload)

validate_boolean_conditionals ¶

validate_boolean_conditionals(
    required_labels: tuple[str, str] | None = None,
) -> None

Raise if any conditional gateway does not have a complete two-branch split.

If required_labels is provided, the gateway must have exactly those two branch labels. If it is omitted, the gateway only needs two distinct, non-empty condition edges.

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

def validate_boolean_conditionals(
    self,
    required_labels: tuple[str, str] | None = None,
) -> None:
    """Raise if any conditional gateway does not have a complete two-branch split.

    If `required_labels` is provided, the gateway must have exactly those two
    branch labels. If it is omitted, the gateway only needs two distinct,
    non-empty condition edges.
    """
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    self.Design.validate_boolean_conditionals(required_labels=required_labels)

validate_connection_rules ¶

validate_connection_rules(
    *,
    logic_tile_types: tuple[str, ...] = (
        "conditionalGatewayNode",
    ),
    allow_dangling_tile_types: tuple[str, ...] = (
        "endNode",
    ),
    required_logic_outputs: int = 2
) -> None

Raise when connection rules are violated.

This is a stricter topological check than validate_boolean_conditionals and is intended to answer whether the workflow graph is structurally valid.

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

def validate_connection_rules(
    self,
    *,
    logic_tile_types: tuple[str, ...] = ("conditionalGatewayNode",),
    allow_dangling_tile_types: tuple[str, ...] = ("endNode",),
    required_logic_outputs: int = 2,
) -> None:
    """Raise when connection rules are violated.

    This is a stricter topological check than ``validate_boolean_conditionals``
    and is intended to answer whether the workflow graph is structurally valid.
    """
    self._sync_design_from_legacy_fields()
    assert self.Design is not None
    self.Design.validate_connection_rules(
        logic_tile_types=logic_tile_types,
        allow_dangling_tile_types=allow_dangling_tile_types,
        required_logic_outputs=required_logic_outputs,
    )

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."
        ),
    )

DomoWorkflow_Design `dataclass` ¶

DomoWorkflow_Design(
    schema_version: int = 2,
    tiles: list[DomoWorkflow_Tile] = list(),
    edges: list[DomoWorkflow_Edge] = list(),
    layouts: list[dict[str, Any]] = list(),
    inputs: dict[str, DomoWorkflow_Parameter] = dict(),
    outputs: dict[str, DomoWorkflow_Parameter] = dict(),
    raw: dict = dict(),
)

Workflow design manager for graph elements and topology validation.

This mirrors the sub-entity composition style used by other entities: design details live here while higher-level model/definition concerns remain in the parent class.