definition

definition

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