manager

manager

DomoDataflow Actions Manager

This module provides the DomoDataflow_Actions manager class for handling dataflow actions with computed properties for analysis and filtering.

DomoDataflow_Actions dataclass

DomoDataflow_Actions(
    auth: DomoAuth,
    dataflow: DomoDataflowProtocol,
    dataflow_id: str,
    actions: list[DomoDataflow_Action_Base] = list(),
)

Manager class for dataflow actions.

This class wraps the actions list from a dataflow and provides computed properties for filtering and analysis.

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	list[DomoDataflow_Action_Base]	List of action objects

Example

dataflow = await DomoDataflow.get_by_id(auth=auth, dataflow_id=123) await dataflow.Actions.get() print(f"Has data science tiles: {dataflow.Actions.has_datascience_tiles}") print(f"Input datasets: {dataflow.Actions.input_datasets}")

action_type_counts property

action_type_counts: dict[str, int]

Get count of actions by action type.

Returns:

Type	Description
dict[str, int]	Dictionary mapping action_type to count

Example

counts = dataflow.Actions.action_type_counts for action_type, count in counts.items(): ... print(f"{action_type}: {count}")

column_relationships property

column_relationships: set[ColumnRelationship]

Aggregate column-level lineage across all actions.

Collects ColumnRelationships from every action's column_relationships property and returns the union.

Example

rels = dataflow.Actions.column_relationships print(f"Found {len(rels)} column relationships")

datascience_tiles property

datascience_tiles: list[DomoDataflow_Action_Base]

Get list of data science actions.

Returns:

Type	Description
list[DomoDataflow_Action_Base]	Filtered list of data science actions

Example

ds_tiles = dataflow.Actions.datascience_tiles print(f"Found {len(ds_tiles)} data science tiles")

disabled_actions property

disabled_actions: list[DomoDataflow_Action_Base]

Get list of disabled actions.

Returns:

Type	Description
list[DomoDataflow_Action_Base]	Filtered list of disabled actions

Example

disabled = dataflow.Actions.disabled_actions print(f"Found {len(disabled)} disabled actions")

has_datascience_tiles property

has_datascience_tiles: bool

Check if the dataflow has any data science tiles.

Returns:

Type	Description
bool	True if any action is a data science tile, False otherwise

Example

if dataflow.Actions.has_datascience_tiles: ... print("This dataflow uses data science operations")

input_datasets property

input_datasets: list[dict[str, Any]]

Get list of input datasets (LoadFromVault actions).

Returns:

Type	Description
list[dict[str, Any]]	List of dicts with 'id' and 'name' keys for each input dataset

Example

for ds in dataflow.Actions.input_datasets: ... print(f"Input: {ds['name']} ({ds['id']})")

output_datasets property

output_datasets: list[dict[str, Any]]

Get list of output datasets (PublishToVault/WriteToVault actions).

Returns:

Type	Description
list[dict[str, Any]]	List of dicts with 'id' and 'name' keys for each output dataset

Example

for ds in dataflow.Actions.output_datasets: ... print(f"Output: {ds['name']} ({ds['id']})")

tile_type_counts property

tile_type_counts: dict[str, int]

Get count of actions by tile type.

Returns:

Type	Description
dict[str, int]	Dictionary mapping tile_type to count

Example

counts = dataflow.Actions.tile_type_counts for tile_type, count in counts.items(): ... print(f"{tile_type}: {count}")

topo_sorted property

topo_sorted: list[DomoDataflow_Action_Base]

Return actions in topological (dependency) order.

Uses the same Kahn's-algorithm sort as convert_entire_workflow_to_sql_tiles so callers get a stable, reproducible ordering that respects dependsOn edges.

Example

for action in dataflow.Actions.topo_sorted: ... print(action.name)

add_action

add_action(obj: dict) -> DomoDataflow_Action_Base

Create a typed action from a raw dict and append it to this manager.

Uses polymorphic dispatch on DomoDataflow_Action_Base.from_dict() so the returned object is the correct concrete subclass.

Parameters:

Name	Type	Description	Default
obj	dict	Raw action dict (e.g. from to_action_dict())	required

Returns:

Type	Description
DomoDataflow_Action_Base	The newly created typed action (already appended to self.actions)

Example

new_dict = existing_pub.to_action_dict(upstream_id=..., output_name=...) new_action = defn.Actions.add_action(new_dict) tile = new_action.to_canvas_tile()

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def add_action(self, obj: dict) -> DomoDataflow_Action_Base:
    """Create a typed action from a raw dict and append it to this manager.

    Uses polymorphic dispatch on DomoDataflow_Action_Base.from_dict() so the
    returned object is the correct concrete subclass.

    Args:
        obj: Raw action dict (e.g. from to_action_dict())

    Returns:
        The newly created typed action (already appended to self.actions)

    Example:
        >>> new_dict = existing_pub.to_action_dict(upstream_id=..., output_name=...)
        >>> new_action = defn.Actions.add_action(new_dict)
        >>> tile = new_action.to_canvas_tile()
    """
    action = DomoDataflow_Action_Base.from_dict(obj, all_actions=self.actions)
    self.actions.append(action)
    return action

convert_entire_workflow_to_sql

convert_entire_workflow_to_sql() -> list[dict[str, Any]]

Alias of convert_entire_workflow_to_sql_tiles().

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def convert_entire_workflow_to_sql(self) -> list[dict[str, Any]]:
    """Alias of convert_entire_workflow_to_sql_tiles()."""
    return self.convert_entire_workflow_to_sql_tiles()

convert_entire_workflow_to_sql_tiles

convert_entire_workflow_to_sql_tiles() -> (
    list[dict[str, Any]]
)

Convert the full workflow into ordered SQL tile approximations.

Returns:

Type	Description
list[dict[str, Any]]	List of tile conversion payloads with SQL, status, and dependencies.

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def convert_entire_workflow_to_sql_tiles(self) -> list[dict[str, Any]]:
    """Convert the full workflow into ordered SQL tile approximations.

    Returns:
        List of tile conversion payloads with SQL, status, and dependencies.
    """
    if not self.actions:
        return []
    return convert_entire_workflow_to_sql_tiles(self.actions)

convert_tile_to_sql

convert_tile_to_sql(
    action_id: str,
) -> dict[str, Any] | None

Convert a single tile/action to SQL with dependency context.

Parameters:

Name	Type	Description	Default
action_id	str	The tile/action ID to convert.	required

Returns:

Type	Description
dict[str, Any] | None	A conversion dict for the matching tile, or None if not found.

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def convert_tile_to_sql(self, action_id: str) -> dict[str, Any] | None:
    """Convert a single tile/action to SQL with dependency context.

    Args:
        action_id: The tile/action ID to convert.

    Returns:
        A conversion dict for the matching tile, or None if not found.
    """
    conversions = self.convert_entire_workflow_to_sql_tiles()
    for row in conversions:
        if row.get("tile_id") == action_id:
            return row
    return None

from_parent classmethod

from_parent(
    parent: DomoDataflowProtocol,
    actions: list[DomoDataflow_Action_Base] | None = None,
) -> DomoDataflow_Actions

Create an Actions manager from a parent dataflow.

Parameters:

Name	Type	Description	Default
parent	DomoDataflowProtocol	Parent DomoDataflow instance	required
actions	list[DomoDataflow_Action_Base] | None	Optional initial list of actions	None

Returns:

Type	Description
DomoDataflow_Actions	DomoDataflow_Actions instance

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

@classmethod
def from_parent(
    cls,
    parent: DomoDataflowProtocol,
    actions: list[DomoDataflow_Action_Base] | None = None,
) -> DomoDataflow_Actions:
    """Create an Actions manager from a parent dataflow.

    Args:
        parent: Parent DomoDataflow instance
        actions: Optional initial list of actions

    Returns:
        DomoDataflow_Actions instance
    """
    return cls(
        auth=parent.auth,
        dataflow=parent,
        dataflow_id=parent.id,
        actions=actions or [],
    )

get async

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

Get actions from the dataflow definition.

This method fetches the latest dataflow definition and populates the actions list.

Parameters:

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

Returns:

Type	Description
list[DomoDataflow_Action_Base]	List of action objects

Example

actions = await dataflow.Actions.get() for action in actions: ... print(f"{action.name}: {action.action_type}")

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

async def get(
    self,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataflow_Action_Base]:
    """Get actions from the dataflow definition.

    This method fetches the latest dataflow definition and populates
    the actions list.

    Args:
        session: Optional httpx client session
        debug_api: Enable debug logging for API calls
        context: Optional RouteContext for the request
        **context_kwargs: Additional context parameters

    Returns:
        List of action objects

    Example:
        >>> actions = await dataflow.Actions.get()
        >>> for action in actions:
        ...     print(f"{action.name}: {action.action_type}")
    """
    # Get the dataflow definition (this populates dataflow.raw)
    # Use the parent dataflow's Definition manager get method

    # If this Actions manager is part of a Definition, use its get method
    # Otherwise, call the definition's get directly
    if hasattr(self.dataflow, "Definition") and self.dataflow.Definition:
        await self.dataflow.Definition.get(
            context=context,
            session=session,  # type: ignore
            debug_api=debug_api,
        )
    else:
        # Fallback: directly fetch and update raw
        from ....routes import dataflow as dataflow_routes

        res = await dataflow_routes.get_dataflow_by_id(
            auth=self.auth,
            dataflow_id=self.dataflow_id,
            context=context,
        )
        if res.is_success:
            self.dataflow.raw = res.response

    # Parse actions from the raw definition
    if self.dataflow.raw and self.dataflow.raw.get("actions"):
        self.actions = [
            DomoDataflow_Action_Base.from_dict(
                action_dict, all_actions=self.actions
            )
            for action_dict in self.dataflow.raw["actions"]
        ]

    return self.actions

get_action_by_id

get_action_by_id(
    action_id: str,
) -> DomoDataflow_Action_Base | None

Get an action by its ID.

Parameters:

Name	Type	Description	Default
action_id	str	ID of the action to find	required

Returns:

Type	Description
DomoDataflow_Action_Base | None	The action object, or None if not found

Example

action = dataflow.Actions.get_action_by_id("abc123") if action: ... print(f"Found action: {action.name}")

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def get_action_by_id(self, action_id: str) -> DomoDataflow_Action_Base | None:
    """Get an action by its ID.

    Args:
        action_id: ID of the action to find

    Returns:
        The action object, or None if not found

    Example:
        >>> action = dataflow.Actions.get_action_by_id("abc123")
        >>> if action:
        ...     print(f"Found action: {action.name}")
    """
    for action in self.actions:
        if action.id == action_id:
            return action
    return None

get_actions_by_type

get_actions_by_type(
    action_type: str,
) -> list[DomoDataflow_Action_Base]

Get all actions of a specific type.

Parameters:

Name	Type	Description	Default
action_type	str	The action type to filter by (e.g., "Filter", "GroupBy")	required

Returns:

Type	Description
list[DomoDataflow_Action_Base]	List of actions matching the type

Example

filters = dataflow.Actions.get_actions_by_type("Filter") print(f"Found {len(filters)} filter actions")

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def get_actions_by_type(self, action_type: str) -> list[DomoDataflow_Action_Base]:
    """Get all actions of a specific type.

    Args:
        action_type: The action type to filter by (e.g., "Filter", "GroupBy")

    Returns:
        List of actions matching the type

    Example:
        >>> filters = dataflow.Actions.get_actions_by_type("Filter")
        >>> print(f"Found {len(filters)} filter actions")
    """
    return [action for action in self.actions if action.action_type == action_type]

get_script_content

get_script_content(action_id: str) -> str | None

Extract script content from Python, R, or SQL script tiles.

Parameters:

Name	Type	Description	Default
action_id	str	ID of the action to extract script from	required

Returns:

Type	Description
str | None	Script content as string, or None if not a script tile or script not found

Example

script = dataflow.Actions.get_script_content("abc123") if script: ... print(script)

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def get_script_content(self, action_id: str) -> str | None:
    """Extract script content from Python, R, or SQL script tiles.

    Args:
        action_id: ID of the action to extract script from

    Returns:
        Script content as string, or None if not a script tile or script not found

    Example:
        >>> script = dataflow.Actions.get_script_content("abc123")
        >>> if script:
        ...     print(script)
    """
    action = self.get_action_by_id(action_id)
    if not action:
        return None

    # Check for direct script field
    if hasattr(action, "script") and action.script:  # type: ignore
        return action.script  # type: ignore

    # Check settings
    if action.settings:
        script = (
            action.settings.get("script")
            or action.settings.get("code")
            or action.settings.get("pythonScript")
            or action.settings.get("rScript")
            or action.settings.get("sql")
            or action.settings.get("query")
            or action.settings.get("sqlScript")
        )
        if script:
            return script

    # Check raw data
    if action.raw:
        return action.raw.get("script")

    return None

remove_action

remove_action(
    action_id: str,
) -> DomoDataflow_Action_Base | None

Remove an action by its ID.

Parameters:

Name	Type	Description	Default
action_id	str	ID of the action to remove.	required

Returns:

Type	Description
DomoDataflow_Action_Base | None	The removed action, or None if not found.

Example::

removed = defn.Actions.remove_action("MergeJoin-abc123")

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def remove_action(self, action_id: str) -> DomoDataflow_Action_Base | None:
    """Remove an action by its ID.

    Args:
        action_id: ID of the action to remove.

    Returns:
        The removed action, or None if not found.

    Example::

        removed = defn.Actions.remove_action("MergeJoin-abc123")
    """
    for i, action in enumerate(self.actions):
        if action.id == action_id:
            return self.actions.pop(i)
    return None

remove_actions_by_type

remove_actions_by_type(
    action_type: str,
) -> list[DomoDataflow_Action_Base]

Remove all actions of a given type.

Idempotent — safe to call repeatedly (returns empty list on second call).

Parameters:

Name	Type	Description	Default
action_type	str	The action type string to remove (e.g., "SplitJoin").	required

Returns:

Type	Description
list[DomoDataflow_Action_Base]	List of removed actions.

Example::

removed = defn.Actions.remove_actions_by_type("SplitJoin")

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def remove_actions_by_type(
    self, action_type: str
) -> list[DomoDataflow_Action_Base]:
    """Remove all actions of a given type.

    Idempotent — safe to call repeatedly (returns empty list on second call).

    Args:
        action_type: The action type string to remove (e.g., "SplitJoin").

    Returns:
        List of removed actions.

    Example::

        removed = defn.Actions.remove_actions_by_type("SplitJoin")
    """
    keep, removed = [], []
    for action in self.actions:
        (removed if action.action_type == action_type else keep).append(action)
    self.actions = keep
    return removed

remove_actions_where

remove_actions_where(
    predicate,
) -> list[DomoDataflow_Action_Base]

Remove all actions matching a predicate.

Idempotent — safe to call repeatedly.

Parameters:

Name	Type	Description	Default
predicate		Callable accepting a DomoDataflow_Action_Base and returning True for actions to remove.	required

Returns:

Type	Description
list[DomoDataflow_Action_Base]	List of removed actions.

Example::

removed = defn.Actions.remove_actions_where(
    lambda a: a.action_type == "PublishToVault" and a.name == "MetaData_Pages_NoStats"
)

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def remove_actions_where(self, predicate) -> list[DomoDataflow_Action_Base]:
    """Remove all actions matching a predicate.

    Idempotent — safe to call repeatedly.

    Args:
        predicate: Callable accepting a DomoDataflow_Action_Base and
            returning True for actions to remove.

    Returns:
        List of removed actions.

    Example::

        removed = defn.Actions.remove_actions_where(
            lambda a: a.action_type == "PublishToVault" and a.name == "MetaData_Pages_NoStats"
        )
    """
    keep, removed = [], []
    for action in self.actions:
        (removed if predicate(action) else keep).append(action)
    self.actions = keep
    return removed

to_erd

to_erd(
    *,
    title: str | None = None,
    trace_from_entity: str | None = None,
    trace_from_column: str | None = None
) -> MermaidERDiagram

Generate a Mermaid ER diagram from column relationships.

Collects column relationships from all actions and converts them to an ER diagram showing column-level lineage between tiles.

Parameters:

Name	Type	Description	Default
title	str | None	Optional diagram title (defaults to dataflow name)	None
trace_from_entity	str | None	If provided, trace only the upstream chain from this entity (e.g., a tile ID)	None
trace_from_column	str | None	If provided, trace only the upstream chain for this specific column (e.g., "Revenue")	None

Returns:

Type	Description
MermaidERDiagram	MermaidERDiagram with entities and relationships

Example

Full dataflow ERD

diagram = dataflow.Actions.to_erd()

Trace a specific column upstream

diagram = dataflow.Actions.to_erd( ... trace_from_entity="publish-1", ... trace_from_column="Revenue", ... )

Source code in src/crew_dcs/classes/DomoDataflow/action/manager.py

def to_erd(
    self,
    *,
    title: str | None = None,
    trace_from_entity: str | None = None,
    trace_from_column: str | None = None,
) -> MermaidERDiagram:
    """Generate a Mermaid ER diagram from column relationships.

    Collects column relationships from all actions and converts them
    to an ER diagram showing column-level lineage between tiles.

    Args:
        title: Optional diagram title (defaults to dataflow name)
        trace_from_entity: If provided, trace only the upstream chain
                          from this entity (e.g., a tile ID)
        trace_from_column: If provided, trace only the upstream chain
                          for this specific column (e.g., "Revenue")

    Returns:
        MermaidERDiagram with entities and relationships

    Example:
        >>> # Full dataflow ERD
        >>> diagram = dataflow.Actions.to_erd()
        >>> # Trace a specific column upstream
        >>> diagram = dataflow.Actions.to_erd(
        ...     trace_from_entity="publish-1",
        ...     trace_from_column="Revenue",
        ... )
    """
    from ....integrations.graphs.mermaid import ColumnRelationshipERConverter

    rels = self.column_relationships
    entity_names = {a.id: a.name or a.id for a in self.actions}
    diagram_title = title or getattr(self, "_dataflow_name", None)
    return ColumnRelationshipERConverter.convert(
        rels,
        entity_names=entity_names,
        title=diagram_title,
        trace_from_entity=trace_from_entity,
        trace_from_column=trace_from_column,
    )