definition

definition

DomoDataflow Definition Manager

This module provides the DomoDataflow_Definition manager class for handling dataflow definitions including actions, triggers, GUI layout, and versioning.

DomoDataflow_Definition dataclass

DomoDataflow_Definition(
    auth: DomoAuth,
    dataflow: DomoDataflowProtocol,
    dataflow_id: str,
    Actions: DomoDataflow_Actions,
    TriggerSettings: DomoTriggerSettings | None = None,
    version_id: int | None = None,
    version_number: int | None = None,
    Canvas: CanvasElements | None = None,
    procedures: list | None = None,
    raw: dict | None = None,
)

Manager class for dataflow definition.

This class wraps the complete dataflow definition including actions, triggers, GUI layout, and version information.

Attributes:

Name	Type	Description
auth	DomoAuth	DomoAuth instance from parent dataflow
dataflow	DomoDataflowProtocol	Reference to parent DomoDataflow
dataflow_id	str	ID of the parent dataflow
Actions	DomoDataflow_Actions	Manager for dataflow actions
TriggerSettings	DomoTriggerSettings | None	Manager for trigger configuration
version_id	int | None	Current version ID
version_number	int | None	Current version number
gui	dict | None	GUI canvas layout configuration
procedures	list | None	List of procedures (legacy)
raw	dict | None	Raw definition data from API

Example

dataflow = await DomoDataflow.get_by_id(auth=auth, dataflow_id=123) await dataflow.Definition.get() print(f"Actions: {len(dataflow.Definition.Actions.actions)}") print(f"Version: {dataflow.Definition.version_number}")

gui property

gui: dict | None

Raw gui dict for backwards compatibility.

canvas_edges

canvas_edges() -> list[tuple[str, str]]

Return (source_id, target_id) pairs derived from action dependsOn.

Cross-domain join: needs tile IDs from Canvas and dependsOn from Actions. Owned here — not delegated to either sub-manager.

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

def canvas_edges(self) -> list[tuple[str, str]]:
    """Return (source_id, target_id) pairs derived from action dependsOn.

    Cross-domain join: needs tile IDs from Canvas and dependsOn from Actions.
    Owned here — not delegated to either sub-manager.
    """
    if not self.Canvas:
        return []
    tile_ids = {t.id for t in self.Canvas.tiles}
    return [
        (dep_id, action.id)
        for action in self.Actions.actions
        for dep_id in (action.depends_on or [])
        if dep_id in tile_ids and action.id in tile_ids
    ]

canvas_layout

canvas_layout() -> list[dict]

Return all canvas elements as normalised dicts.

Delegates to CanvasElements.layout() but enriches Tile labels with action names from the Actions manager.

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

def canvas_layout(self) -> list[dict]:
    """Return all canvas elements as normalised dicts.

    Delegates to CanvasElements.layout() but enriches Tile labels with
    action names from the Actions manager.
    """
    if not self.Canvas:
        return []
    action_names: dict[str, str] = {
        a.id: a.name for a in self.Actions.actions if a.id and a.name
    }
    layout = self.Canvas.layout()
    for el in layout:
        if el["type"] == "Tile" and el["id"] in action_names:
            el["label"] = action_names[el["id"]]
    return layout

from_parent classmethod

from_parent(
    parent: DomoDataflowProtocol,
    actions: list | None = None,
    trigger_settings: dict | None = None,
) -> DomoDataflow_Definition

Create a Definition manager from a parent dataflow.

Parameters:

Name	Type	Description	Default
parent	DomoDataflowProtocol	Parent DomoDataflow instance	required
actions	list | None	Optional initial list of actions	None
trigger_settings	dict | None	Optional trigger settings dict	None

Returns:

Type	Description
DomoDataflow_Definition	DomoDataflow_Definition instance

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

@classmethod
def from_parent(
    cls,
    parent: DomoDataflowProtocol,
    actions: list | None = None,
    trigger_settings: dict | None = None,
) -> DomoDataflow_Definition:
    """Create a Definition manager from a parent dataflow.

    Args:
        parent: Parent DomoDataflow instance
        actions: Optional initial list of actions
        trigger_settings: Optional trigger settings dict

    Returns:
        DomoDataflow_Definition instance
    """
    # Create Actions manager
    actions_manager = DomoDataflow_Actions.from_parent(
        parent=parent, actions=actions
    )

    # Create TriggerSettings if provided
    trigger_mgr = None
    if trigger_settings:
        trigger_mgr = DomoTriggerSettings.from_parent(
            parent=parent, obj=trigger_settings
        )

    return cls(
        auth=parent.auth,
        dataflow=parent,
        dataflow_id=parent.id,
        Actions=actions_manager,
        TriggerSettings=trigger_mgr,
        version_id=parent.version_id,
        version_number=parent.version_number,
        Canvas=CanvasElements.from_dict(
            parent.raw.get("gui") if parent.raw else None
        ),
        procedures=parent.raw.get("procedures") if parent.raw else None,
        raw=parent.raw,
    )

get async

get(
    debug_api: bool = False,
    return_raw: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs
)

Fetch and refresh the dataflow definition.

This method fetches the latest dataflow definition from the API and updates all definition components (actions, triggers, GUI, etc.).

Parameters:

Name	Type	Description	Default
debug_api	bool	Enable debug logging for API calls	False
return_raw	bool	Return raw API response instead of updating	False
session	AsyncClient | None	Optional httpx client session	None
context	RouteContext | None	Optional RouteContext for the request	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	Self if return_raw=False, ResponseGetData if return_raw=True

Example

await dataflow.Definition.get() print(f"Loaded {len(dataflow.Definition.Actions.actions)} actions")

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

async def get(
    self,
    debug_api: bool = False,
    return_raw: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Fetch and refresh the dataflow definition.

    This method fetches the latest dataflow definition from the API
    and updates all definition components (actions, triggers, GUI, etc.).

    Args:
        debug_api: Enable debug logging for API calls
        return_raw: Return raw API response instead of updating
        session: Optional httpx client session
        context: Optional RouteContext for the request
        **context_kwargs: Additional context parameters

    Returns:
        Self if return_raw=False, ResponseGetData if return_raw=True

    Example:
        >>> await dataflow.Definition.get()
        >>> print(f"Loaded {len(dataflow.Definition.Actions.actions)} actions")
    """
    context = RouteContext.build_context(
        session=session, debug_api=debug_api, **context_kwargs
    )

    res = await dataflow_routes.get_dataflow_by_id(
        auth=self.auth,
        dataflow_id=self.dataflow_id,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        return self

    # Update raw data
    self.raw = res.response
    self.dataflow.raw = res.response

    # Update version info
    self.version_id = res.response.get("versionId")
    self.version_number = res.response.get("versionNumber")
    self.dataflow.version_id = self.version_id
    self.dataflow.version_number = self.version_number

    # Update Canvas and procedures
    self.Canvas = CanvasElements.from_dict(res.response.get("gui"))
    self.procedures = res.response.get("procedures")

    # Update parent dataflow attributes
    self.dataflow.name = res.response.get("name")
    self.dataflow.description = res.response.get("description")
    self.dataflow.owner = res.response.get("owner")

    # Re-parse actions from the new definition
    if self.raw and self.raw.get("actions"):
        self.Actions.actions = [
            DomoDataflow_Action_Base.from_dict(action_dict, all_actions=[])
            for action_dict in self.raw["actions"]
        ]

    # Update trigger settings if present
    if self.raw.get("triggerSettings"):
        self.TriggerSettings = DomoTriggerSettings.from_parent(
            parent=self.dataflow, obj=self.raw["triggerSettings"]
        )

    return self

next_free_position

next_free_position(
    width: int = 224, height: int = 96, padding: int = 32
) -> tuple[int, int]

Return (x, y) for a new element that does not overlap any existing element.

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

def next_free_position(
    self, width: int = 224, height: int = 96, padding: int = 32
) -> tuple[int, int]:
    """Return (x, y) for a new element that does not overlap any existing element."""
    if not self.Canvas:
        return (32, 32)
    return self.Canvas.next_free_position(
        width=width, height=height, padding=padding
    )

optimize_canvas

optimize_canvas(
    add_sections: bool = True,
    save: bool = False,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> CanvasElements

Optimize the canvas layout using topological positioning.

Computes a layered layout from the action dependency graph, applies standard tile colours by action type, and optionally groups tiles into sections by upstream root.

Parameters:

Name	Type	Description	Default
add_sections	bool	Whether to create auto-sections (default True).	True
save	bool	Whether to save the definition after optimizing (default False).	False
debug_api	bool	Enable debug logging for API calls (only used if save=True).	False
session	AsyncClient | None	Optional httpx client session (only used if save=True).	None
context	RouteContext | None	Optional RouteContext for the request (only used if save=True).	None
**context_kwargs		Additional context parameters (only used if save=True).	{}

Returns:

Type	Description
CanvasElements	The modified :class:CanvasElements instance.

Example::

defn.optimize_canvas(add_sections=True)
# or with auto-save:
defn.optimize_canvas(save=True, version_note="Optimized layout")

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

def optimize_canvas(
    self,
    add_sections: bool = True,
    save: bool = False,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> CanvasElements:
    """Optimize the canvas layout using topological positioning.

    Computes a layered layout from the action dependency graph,
    applies standard tile colours by action type, and optionally
    groups tiles into sections by upstream root.

    Args:
        add_sections: Whether to create auto-sections (default True).
        save: Whether to save the definition after optimizing (default False).
        debug_api: Enable debug logging for API calls (only used if save=True).
        session: Optional httpx client session (only used if save=True).
        context: Optional RouteContext for the request (only used if save=True).
        **context_kwargs: Additional context parameters (only used if save=True).

    Returns:
        The modified :class:`CanvasElements` instance.

    Example::

        defn.optimize_canvas(add_sections=True)
        # or with auto-save:
        defn.optimize_canvas(save=True, version_note="Optimized layout")
    """
    from .layout_optimizer import optimize_canvas as _optimize

    canvas = _optimize(self, add_sections=add_sections)

    if save:
        # save() is async so we return a coroutine that the caller awaits
        return self.save(
            debug_api=debug_api,
            session=session,
            version_note="Optimized canvas layout",
            context=context,
            **context_kwargs,
        )

    return canvas

save async

save(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    version_note: str | None = None,
    context: RouteContext | None = None,
    **context_kwargs
)

Write current Canvas + Actions state back to the API.

Builds the full definition payload from the typed managers so callers don't have to hand-craft raw dicts. Equivalent to calling update(new_definition) with a payload assembled from self.Canvas and self.Actions.

Parameters:

Name	Type	Description	Default
debug_api	bool	Enable debug logging for API calls.	False
session	AsyncClient | None	Optional httpx client session.	None
version_note	str | None	Optional version description shown in the Domo UI history (onboardFlowVersion.description). If omitted the existing value (or server default) is preserved.	None
context	RouteContext | None	Optional RouteContext for the request.	None
**context_kwargs		Additional context parameters.	{}

Example::

await defn.save(version_note="Replaced MergeJoin with SplitJoin")

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

async def save(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    version_note: str | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Write current Canvas + Actions state back to the API.

    Builds the full definition payload from the typed managers so callers
    don't have to hand-craft raw dicts. Equivalent to calling
    update(new_definition) with a payload assembled from self.Canvas and
    self.Actions.

    Args:
        debug_api: Enable debug logging for API calls.
        session: Optional httpx client session.
        version_note: Optional version description shown in the Domo UI
            history (``onboardFlowVersion.description``). If omitted the
            existing value (or server default) is preserved.
        context: Optional RouteContext for the request.
        **context_kwargs: Additional context parameters.

    Example::

        await defn.save(version_note="Replaced MergeJoin with SplitJoin")
    """
    if not self.raw:
        raise RuntimeError(
            "Cannot save: Definition has not been fetched yet (call get() first)"
        )

    new_def = dict(self.raw)
    new_def["actions"] = [a.raw for a in self.Actions.actions if a.raw]
    if self.Canvas:
        new_def["gui"] = self.Canvas.to_dict()

    if version_note is not None:
        new_def.setdefault("onboardFlowVersion", {})
        new_def["onboardFlowVersion"]["description"] = version_note

    return await self.update(
        new_def,
        debug_api=debug_api,
        session=session,
        context=context,
        **context_kwargs,
    )

save_and_export_svg async

save_and_export_svg(
    svg_path: str,
    *,
    title: str | None = None,
    version_note: str | None = None,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> str

Save the definition and export an SVG rendering in one step.

Convenience method that calls :meth:save then :meth:to_svg and writes the SVG to svg_path.

Parameters:

Name	Type	Description	Default
svg_path	str	File path to write the SVG to.	required
title	str | None	Optional title for the SVG (defaults to dataflow name).	None
version_note	str | None	Optional version description for the Domo UI history.	None
debug_api	bool	Enable debug logging for API calls.	False
session	AsyncClient | None	Optional httpx client session.	None
context	RouteContext | None	Optional RouteContext for the request.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Type	Description
str	The absolute path to the written SVG file.

Example::

await defn.save_and_export_svg("/tmp/dataflow-509.svg",
    version_note="Replaced MergeJoin with SplitJoin")

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

async def save_and_export_svg(
    self,
    svg_path: str,
    *,
    title: str | None = None,
    version_note: str | None = None,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> str:
    """Save the definition and export an SVG rendering in one step.

    Convenience method that calls :meth:`save` then :meth:`to_svg`
    and writes the SVG to *svg_path*.

    Args:
        svg_path: File path to write the SVG to.
        title: Optional title for the SVG (defaults to dataflow name).
        version_note: Optional version description for the Domo UI history.
        debug_api: Enable debug logging for API calls.
        session: Optional httpx client session.
        context: Optional RouteContext for the request.
        **context_kwargs: Additional context parameters.

    Returns:
        The absolute path to the written SVG file.

    Example::

        await defn.save_and_export_svg("/tmp/dataflow-509.svg",
            version_note="Replaced MergeJoin with SplitJoin")
    """
    await self.save(
        debug_api=debug_api,
        session=session,
        version_note=version_note,
        context=context,
        **context_kwargs,
    )
    svg = self.to_svg(title=title)
    import os

    from crew_dcs.utils.files import upsert_folder

    svg_path = os.path.normpath(svg_path)
    upsert_folder(os.path.dirname(svg_path))
    with open(svg_path, "w") as f:
        f.write(svg)
    return svg_path

to_svg

to_svg(title: str | None = None) -> str

Render the canvas as an SVG string (call get() first).

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

def to_svg(self, title: str | None = None) -> str:
    """Render the canvas as an SVG string (call get() first)."""
    from .canvas import render_svg

    return render_svg(
        layout=self.canvas_layout(),
        edges=self.canvas_edges(),
        title=title or (self.dataflow.name if self.dataflow else "Dataflow Canvas"),
    )

update async

update(
    new_definition: dict,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update the dataflow definition.

This method updates the dataflow definition on the server and then refreshes the local state.

Parameters:

Name	Type	Description	Default
new_definition	dict	New definition dictionary	required
debug_api	bool	Enable debug logging for API calls	False
debug_num_stacks_to_drop	int	Stack frames to drop in debug logging	2
session	AsyncClient | None	Optional httpx client session	None
context	RouteContext | None	Optional RouteContext for the request	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
	Self with updated definition

Example

new_def = {"name": "Updated Name", "actions": [...]} await dataflow.Definition.update(new_def)

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

async def update(
    self,
    new_definition: dict,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update the dataflow definition.

    This method updates the dataflow definition on the server and
    then refreshes the local state.

    Args:
        new_definition: New definition dictionary
        debug_api: Enable debug logging for API calls
        debug_num_stacks_to_drop: Stack frames to drop in debug logging
        session: Optional httpx client session
        context: Optional RouteContext for the request
        **context_kwargs: Additional context parameters

    Returns:
        Self with updated definition

    Example:
        >>> new_def = {"name": "Updated Name", "actions": [...]}
        >>> await dataflow.Definition.update(new_def)
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )

    await dataflow_routes.update_dataflow_definition(
        auth=self.auth,
        dataflow_id=self.dataflow_id,
        dataflow_definition=new_definition,
        context=context,
    )

    return await self.get(return_raw=False, session=session, debug_api=debug_api)