design

design

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.

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/design.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."""
    true_edge = self.add_edge(
        gateway_id,
        true_target_id,
        title=true_label,
        condition_type=condition_type,
    )
    false_edge = self.add_edge(
        gateway_id,
        false_target_id,
        title=false_label,
        condition_type=condition_type,
    )
    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/design.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."""
    edge = DomoWorkflow_Edge.create(
        source_id=source_id,
        target_id=target_id,
        title=title,
        condition_type=condition_type,
    )
    self.edges.append(edge)
    return edge

add_input

add_input(param: DomoWorkflow_Parameter) -> None

Add a parameter to the workflow-level schema inputs.

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

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

add_output

add_output(param: DomoWorkflow_Parameter) -> None

Add a parameter to the workflow-level schema outputs.

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

def add_output(self, param: DomoWorkflow_Parameter) -> None:
    """Add a parameter to the workflow-level 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/design.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)."""
    if x is not None:
        tile.x = x
    if y is not None:
        tile.y = y
    elif x is None and y is None:
        if self.tiles:
            last = self.tiles[-1]
            tile.x = last.x
            tile.y = last.y + 150
        else:
            tile.x = 0.0
            tile.y = 0.0
    tile.index = len(self.tiles)
    self.tiles.append(tile)
    return tile

empty classmethod

empty() -> DomoWorkflow_Design

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

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

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

ensure_edge

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

Idempotently add an edge; no-op if an equivalent edge already exists.

Returns the created edge, or None if nothing needed to change.

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

def ensure_edge(
    self,
    source_id: str,
    target_id: str,
    title: str | None = None,
    condition_type: str | None = None,
) -> DomoWorkflow_Edge | None:
    """Idempotently add an edge; no-op if an equivalent edge already exists.

    Returns the created edge, or None if nothing needed to change.
    """
    if self.has_edge(source_id, target_id, label=title):
        return None
    return self.add_edge(
        source_id, target_id, title=title, condition_type=condition_type
    )

find_tile

find_tile(
    *, title: str, tile_type: str | None = None
) -> DomoWorkflow_Tile

Find a single tile by title (and optionally tile_type).

Raises ValueError unless exactly one tile matches — use this instead of hand-rolled list comprehensions when writing targeted graph repairs.

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

def find_tile(
    self,
    *,
    title: str,
    tile_type: str | None = None,
) -> DomoWorkflow_Tile:
    """Find a single tile by title (and optionally tile_type).

    Raises ValueError unless exactly one tile matches — use this instead of
    hand-rolled list comprehensions when writing targeted graph repairs.
    """
    matches = [
        tile
        for tile in self.tiles
        if tile.title == title
        and (tile_type is None or tile.tile_type == tile_type)
    ]
    if len(matches) != 1:
        raise ValueError(
            f"expected exactly 1 tile for title={title!r} type={tile_type!r}, "
            f"found {len(matches)}"
        )
    return matches[0]

from_raw classmethod

from_raw(raw: dict[str, Any]) -> DomoWorkflow_Design

Build a design object from workflow savedJson payload.

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

@classmethod
def from_raw(cls, raw: dict[str, Any]) -> DomoWorkflow_Design:
    """Build a design object from workflow savedJson payload."""
    edge_types = {"defaultEdge", "conditionEdge"}
    tiles: list[DomoWorkflow_Tile] = []
    edges: list[DomoWorkflow_Edge] = []
    tile_index = 0

    for element in raw.get("designElements") or []:
        elem_type = element.get("type", "")
        if elem_type in edge_types:
            edges.append(DomoWorkflow_Edge.from_dict(element))
        else:
            tiles.append(DomoWorkflow_Tile.from_dict(element, index=tile_index))
            tile_index += 1

    schema = raw.get("schema") or {}
    inputs = {
        k: DomoWorkflow_Parameter.from_dict(v)
        for k, v in (schema.get("inputs") or {}).items()
    }
    outputs = {
        k: DomoWorkflow_Parameter.from_dict(v)
        for k, v in (schema.get("outputs") or {}).items()
    }

    return cls(
        schema_version=raw.get("version", 2),
        tiles=tiles,
        edges=edges,
        layouts=raw.get("dataList") or [],
        inputs=inputs,
        outputs=outputs,
        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.

Source code in src/crew_dcs/classes/DomoWorkflow/design.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."""
    tile_by_id = {tile.id: tile for tile in self.tiles}
    outgoing_by_source: dict[str, list[DomoWorkflow_Edge]] = {
        tile.id: [] for tile in self.tiles
    }

    errors: list[str] = []
    for edge in self.edges:
        if edge.source not in tile_by_id:
            errors.append(
                f"edge '{edge.id}' source '{edge.source}' does not reference an existing tile"
            )
            continue
        if edge.target not in tile_by_id:
            errors.append(
                f"edge '{edge.id}' target '{edge.target}' does not reference an existing tile"
            )
            continue
        outgoing_by_source[edge.source].append(edge)

    for tile in self.tiles:
        outgoing = outgoing_by_source.get(tile.id, [])
        tile_label = tile.title or tile.id

        if tile.tile_type not in allow_dangling_tile_types and len(outgoing) == 0:
            errors.append(
                f"tile '{tile_label}' ({tile.tile_type}) has no outgoing connection"
            )

        if (
            tile.tile_type in logic_tile_types
            and len(outgoing) != required_logic_outputs
        ):
            errors.append(
                f"logic tile '{tile_label}' ({tile.tile_type}) must have exactly "
                f"{required_logic_outputs} outgoing connections (found {len(outgoing)})"
            )

    return errors

has_edge

has_edge(
    source_id: str,
    target_id: str,
    *,
    label: str | None = None
) -> bool

Return True if an edge from source_id to target_id already exists.

When label is given, also requires the edge's title to match.

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

def has_edge(
    self,
    source_id: str,
    target_id: str,
    *,
    label: str | None = None,
) -> bool:
    """Return True if an edge from source_id to target_id already exists.

    When label is given, also requires the edge's title to match.
    """
    return any(
        edge.source == source_id
        and edge.target == target_id
        and (label is None or edge.title == label)
        for edge in self.edges
    )

prune_orphan_tiles

prune_orphan_tiles(
    *, tile_types: tuple[str, ...] = ("EMPTY",)
) -> list[str]

Remove tiles of the given type(s) that have no incoming or outgoing edges.

Returns the ids of removed tiles (empty list if nothing was pruned).

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

def prune_orphan_tiles(
    self,
    *,
    tile_types: tuple[str, ...] = ("EMPTY",),
) -> list[str]:
    """Remove tiles of the given type(s) that have no incoming or outgoing edges.

    Returns the ids of removed tiles (empty list if nothing was pruned).
    """
    connected_ids = {edge.target for edge in self.edges} | {
        edge.source for edge in self.edges
    }

    keep: list[DomoWorkflow_Tile] = []
    removed: list[str] = []
    for tile in self.tiles:
        if tile.tile_type in tile_types and tile.id not in connected_ids:
            removed.append(tile.id)
            continue
        keep.append(tile)

    if removed:
        self.tiles = keep
        for idx, tile in enumerate(self.tiles):
            tile.index = idx

    return removed

rewire_gateway_branches

rewire_gateway_branches(
    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]

Replace ALL outgoing edges of a gateway with a fresh True/False pair.

Idempotent to call repeatedly — safe whether the gateway currently has no branches, partial/incorrect branches, or already-correct branches.

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

def rewire_gateway_branches(
    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]:
    """Replace ALL outgoing edges of a gateway with a fresh True/False pair.

    Idempotent to call repeatedly — safe whether the gateway currently has
    no branches, partial/incorrect branches, or already-correct branches.
    """
    self.edges = [edge for edge in self.edges if edge.source != gateway_id]
    return self.add_boolean_branch_edges(
        gateway_id,
        true_target_id,
        false_target_id,
        true_label=true_label,
        false_label=false_label,
        condition_type=condition_type,
    )

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 preflight-gated workflow with an email notification branch.

Source code in src/crew_dcs/classes/DomoWorkflow/design.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 preflight-gated workflow with an email notification branch."""
    from .tile.code_engine import DomoWorkflow_Tile_CodeEngine
    from .tile.conditional import DomoWorkflow_Tile_Conditional
    from .tile.end import DomoWorkflow_Tile_End
    from .tile.misc import DomoWorkflow_Tile_UserTask
    from .tile.start import DomoWorkflow_Tile_Start

    if not success_functions:
        raise ValueError("success_functions must contain at least one function")

    inputs_by_function = inputs_by_function or {}
    failure_notification_emails = failure_notification_emails or []

    start_tile = DomoWorkflow_Tile_Start(
        id="start",
        tile_type="rootNode",
        index=0,
        x=start_x,
        y=start_y,
        title=start_title,
    )

    tiles: list[DomoWorkflow_Tile] = [start_tile]
    current_y = start_y + y_step

    preflight_github = DomoWorkflow_Tile_CodeEngine.create(
        title="get_github_pat",
        package_id=package_id,
        function_name="get_github_pat",
        version=package_version,
        tile_id=str(uuid.uuid4()),
        x=start_x,
        y=current_y,
        inputs=inputs_by_function.get("get_github_pat", []),
    )
    tiles.append(preflight_github)
    current_y += y_step

    github_gateway = DomoWorkflow_Tile_Conditional(
        id=str(uuid.uuid4()),
        tile_type="conditionalGatewayNode",
        index=len(tiles),
        x=start_x,
        y=current_y,
        title="GitHub PAT ready",
        inclusive=False,
    )
    tiles.append(github_gateway)
    current_y += y_step

    preflight_domo = DomoWorkflow_Tile_CodeEngine.create(
        title="get_domo_token",
        package_id=package_id,
        function_name="get_domo_token",
        version=package_version,
        tile_id=str(uuid.uuid4()),
        x=start_x,
        y=current_y,
        inputs=inputs_by_function.get("get_domo_token", []),
    )
    tiles.append(preflight_domo)
    current_y += y_step

    domo_gateway = DomoWorkflow_Tile_Conditional(
        id=str(uuid.uuid4()),
        tile_type="conditionalGatewayNode",
        index=len(tiles),
        x=start_x,
        y=current_y,
        title="Domo token ready",
        inclusive=False,
    )
    tiles.append(domo_gateway)
    current_y += y_step

    success_tiles: list[DomoWorkflow_Tile_CodeEngine] = []
    for fn_name in success_functions:
        tile = DomoWorkflow_Tile_CodeEngine.create(
            title=fn_name,
            package_id=package_id,
            function_name=fn_name,
            version=package_version,
            tile_id=str(uuid.uuid4()),
            x=start_x,
            y=current_y,
            inputs=inputs_by_function.get(fn_name, []),
        )
        success_tiles.append(tile)
        tiles.append(tile)
        current_y += y_step

    notification_tile = DomoWorkflow_Tile_UserTask(
        id=str(uuid.uuid4()),
        tile_type="userTaskNode",
        index=len(tiles),
        x=start_x,
        y=current_y,
        title=notification_title,
        config_type="email",
        email={
            "emailAddresses": failure_notification_emails,
            "subject": "Domo fileset workflow needs attention",
            "body": [
                {"type": "paragraph", "children": [{"text": notification_message}]}
            ],
        },
    )
    tiles.append(notification_tile)
    current_y += y_step

    end_tile = DomoWorkflow_Tile_End(
        id="end",
        tile_type="endNode",
        index=len(tiles),
        x=start_x,
        y=current_y,
        title=end_title,
    )
    tiles.append(end_tile)

    self.tiles = tiles
    self.edges = []

    for idx, tile in enumerate(self.tiles):
        tile.index = idx

    self.add_edge("start", preflight_github.id)
    self.add_edge(preflight_github.id, github_gateway.id)
    self.add_boolean_branch_edges(
        github_gateway.id,
        preflight_domo.id,
        notification_tile.id,
    )
    self.add_edge(preflight_domo.id, domo_gateway.id)
    self.add_boolean_branch_edges(
        domo_gateway.id,
        success_tiles[0].id,
        notification_tile.id,
    )

    prev_id = success_tiles[0].id
    for tile in success_tiles[1:]:
        self.add_edge(prev_id, tile.id)
        prev_id = tile.id

    self.add_edge(prev_id, end_tile.id)
    self.add_edge(notification_tile.id, end_tile.id)

    self.validate_boolean_conditionals(required_labels=("True", "False"))

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 tiles/edges with a start -> CodeEngine chain -> end.

Source code in src/crew_dcs/classes/DomoWorkflow/design.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 tiles/edges with a start -> CodeEngine chain -> end."""
    from .tile.code_engine import DomoWorkflow_Tile_CodeEngine
    from .tile.end import DomoWorkflow_Tile_End
    from .tile.start import DomoWorkflow_Tile_Start

    if not function_names:
        raise ValueError("function_names must contain at least one function")

    inputs_by_function = inputs_by_function or {}

    start_id = "start"
    end_id = "end"

    start_tile = DomoWorkflow_Tile_Start(
        id=start_id,
        tile_type="rootNode",
        index=0,
        x=start_x,
        y=start_y,
        title=start_title,
    )

    tiles = [start_tile]
    curr_y = start_y + y_step

    for fn_name in function_names:
        tile_id = str(uuid.uuid4())
        tile = DomoWorkflow_Tile_CodeEngine.create(
            title=fn_name,
            package_id=package_id,
            function_name=fn_name,
            version=package_version,
            tile_id=tile_id,
            x=start_x,
            y=curr_y,
            inputs=inputs_by_function.get(fn_name, []),
        )
        tiles.append(tile)
        curr_y += y_step

    end_tile = DomoWorkflow_Tile_End(
        id=end_id,
        tile_type="endNode",
        index=len(tiles),
        x=start_x,
        y=curr_y,
        title=end_title,
    )
    tiles.append(end_tile)

    self.tiles = tiles
    self.edges = []

    for idx, tile in enumerate(self.tiles):
        tile.index = idx

    prev_id = start_id
    for tile in self.tiles[1:]:
        self.add_edge(prev_id, tile.id)
        prev_id = tile.id

to_dict

to_dict() -> dict[str, Any]

Serialize the design payload expected by Domo APIs.

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

def to_dict(self) -> dict[str, Any]:
    """Serialize the design payload expected by Domo APIs."""
    design_elements = [t.to_dict() for t in self.tiles] + [
        e.to_dict() for e in self.edges
    ]
    return {
        "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": self.schema_version,
        "dataList": self.layouts,
        "designElements": design_elements,
    }

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.

Source code in src/crew_dcs/classes/DomoWorkflow/design.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."""
    edges_by_source: dict[str, list[DomoWorkflow_Edge]] = {}
    for edge in self.edges:
        edges_by_source.setdefault(edge.source, []).append(edge)

    errors: list[str] = []
    for tile in self.tiles:
        if tile.tile_type != "conditionalGatewayNode":
            continue

        branch_edges = [
            edge
            for edge in edges_by_source.get(tile.id, [])
            if edge.edge_type == "conditionEdge"
        ]
        labels = {edge.title for edge in branch_edges}
        if required_labels is None:
            label_ok = (
                len(branch_edges) == 2
                and len(labels) == 2
                and None not in labels
                and "" not in labels
            )
        else:
            label_ok = len(branch_edges) == 2 and labels == set(required_labels)

        if not label_ok:
            errors.append(
                f"conditional gateway '{tile.title or tile.id}' must have exactly two labeled branches"
            )

        if any(not edge.target for edge in branch_edges):
            errors.append(
                f"conditional gateway '{tile.title or tile.id}' has an unconnected branch"
            )

    if errors:
        raise ValueError("; ".join(errors))

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.

Source code in src/crew_dcs/classes/DomoWorkflow/design.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."""
    errors = self.get_connection_validation_errors(
        logic_tile_types=logic_tile_types,
        allow_dangling_tile_types=allow_dangling_tile_types,
        required_logic_outputs=required_logic_outputs,
    )
    if errors:
        raise ValueError("; ".join(errors))

validate_with_domo async

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

Validate this design with Domo's official workflow validator endpoint.

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

async def validate_with_domo(
    self,
    *,
    auth: DomoAuth,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """Validate this design with Domo's official workflow validator endpoint."""
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )
    return await workflow_routes.validate_workflow_model(
        auth=auth,
        body=self.to_dict(),
        context=context,
    )