action

action

DomoDataflow Action Types

This module provides a registration pattern for Magic ETL v2 action types. Action classes are organized by their category in the Domo ETL UI sidebar.

The folder structure matches Domo's UI categories: - datasets: Input/Output actions (LoadFromVault, PublishToVault, etc.) - utility: General utility actions (SQL, SelectValues, Constant, etc.) - filter: Filter and data cleaning (Filter, Unique) - text: Text manipulation (ConcatFields, ReplaceString, etc.) - dates_numbers: Date and numeric calculations - combine: Joining and combining data (MergeJoin, UnionAll, SplitJoin) - aggregate: Aggregation actions (GroupBy, WindowAction) - pivot: Pivot and unpivot operations - scripting: Custom code execution (PythonEngine) - data_science: ML and data science (MLInferenceAction, UserDefined) - ai_services: AI-powered actions (TextGeneration)

Usage

from crew_dcs.classes.DomoDataflow.action import ( DomoDataflow_Action_Base, DomoDataflow_Action_LoadFromVault, DomoDataflow_Action_SQL, )

Create typed action from API response dict (polymorphic dispatch on base)

action = DomoDataflow_Action_Base.from_dict(raw_action_dict)

Or use the manager to create + append in one call

new_action = defn.Actions.add_action(new_action_dict)

Type-check the result

if isinstance(action, DomoDataflow_Action_SQL): print(f"SQL: {action.sql}")

DomoDataflow_ActionResult dataclass

DomoDataflow_ActionResult(
    id: str,
    type: str = None,
    name: str = None,
    is_success: bool = None,
    rows_processed: int = None,
    begin_time: datetime = None,
    end_time: datetime = None,
    duration_in_sec: int = None,
)

Result of an action execution from dataflow history.

DomoDataflow_Action_Base dataclass

DomoDataflow_Action_Base(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
)

Base class for all dataflow action types.

All specific action types should inherit from this class and use the @register_action_type decorator.

Common fields present in all action types

• id: Unique identifier for the action
• action_type: The type string (e.g., "LoadFromVault")
• tile_type: The category/type of tile (e.g., 'filter', 'aggregate', 'pivot')
• name: Display name of the action (tile name)
• depends_on: List of action IDs this action depends on
• disabled: Whether the action is disabled
• gui: GUI positioning data
• settings: Action-specific settings
• raw: Original API response dict

column_relationships property

column_relationships: set[ColumnRelationship]

Column-level lineage edges for this action.

Base implementation returns an empty set. Action subclasses override this to produce typed ColumnRelationships based on their specific column configuration (fields, groups, keys, etc.).

Direction convention (matches ColumnRelationship): from = downstream (this action's output) to = upstream (input / parent action or dataset)

Returns:

Type	Description
set[ColumnRelationship]	Set of ColumnRelationship edges

is_datascience_tile property

is_datascience_tile: bool

Check if this action is a data science tile.

Returns True if the action type is in DATA_SCIENCE_ACTION_TYPES or if the tile category is 'data_science'.

Returns:

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

sql_conversion_status property

sql_conversion_status: str

Return SQL conversion coverage status for this tile.

add_note

add_note(text: str) -> None

Append a tile note to this action's notes list.

Domo stores tile notes as objects with null coordinate placeholders and a body key. This method appends a new note in that format so it appears in the Domo ETL UI when hovering over the tile.

Idempotent: if a note with the same body text already exists, the duplicate is silently skipped (safe to call on every run).

Parameters:

Name	Type	Description	Default
text	str	The note text to add.	required

Example::

action.add_note("⚠️ Suspicious aggregate — always yields 1")

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

def add_note(self, text: str) -> None:
    """Append a tile note to this action's notes list.

    Domo stores tile notes as objects with null coordinate placeholders
    and a ``body`` key.  This method appends a new note in that format
    so it appears in the Domo ETL UI when hovering over the tile.

    Idempotent: if a note with the same ``body`` text already exists,
    the duplicate is silently skipped (safe to call on every run).

    Args:
        text: The note text to add.

    Example::

        action.add_note("⚠️ Suspicious aggregate — always yields 1")
    """
    if self.raw is None:
        raise RuntimeError(
            "Cannot add note: action has no raw dict (was it created from_dict?)"
        )
    if "notes" not in self.raw or not isinstance(self.raw["notes"], list):
        self.raw["notes"] = []

    # Skip if this exact note body already exists (idempotent on re-runs)
    if any(n.get("body") == text for n in self.raw["notes"]):
        return

    self.raw["notes"].append(
        {"x1": None, "y1": None, "x2": None, "y2": None, "body": text}
    )

convert_to_sql

convert_to_sql(
    *,
    previous_step: str = "source_dataset",
    input_steps: list[str] | None = None,
    input_names: list[str] | None = None
) -> str

Convert this tile/action to a SQL approximation.

Parameters:

Name	Type	Description	Default
previous_step	str	Fallback upstream SQL alias when no dependencies exist.	'source_dataset'
input_steps	list[str] | None	Optional ordered upstream step aliases.	None
input_names	list[str] | None	Optional ordered upstream human-readable names.	None

Returns:

Type	Description
str	SQL approximation string for this tile.

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

def convert_to_sql(
    self,
    *,
    previous_step: str = "source_dataset",
    input_steps: list[str] | None = None,
    input_names: list[str] | None = None,
) -> str:
    """Convert this tile/action to a SQL approximation.

    Args:
        previous_step: Fallback upstream SQL alias when no dependencies exist.
        input_steps: Optional ordered upstream step aliases.
        input_names: Optional ordered upstream human-readable names.

    Returns:
        SQL approximation string for this tile.
    """
    from .sql_converter import tile_to_sql

    return tile_to_sql(
        self,
        previous_step=previous_step,
        input_steps=input_steps,
        input_names=input_names,
    )

dedupe_notes

dedupe_notes() -> int

Remove duplicate notes with the same body text.

Returns:

Type	Description
int	The number of duplicate notes removed.

Example::

removed = action.dedupe_notes()

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

def dedupe_notes(self) -> int:
    """Remove duplicate notes with the same ``body`` text.

    Returns:
        The number of duplicate notes removed.

    Example::

        removed = action.dedupe_notes()
    """
    if self.raw is None or "notes" not in self.raw:
        return 0
    seen: set[str] = set()
    unique: list[dict] = []
    for note in self.raw["notes"]:
        body = note.get("body", "")
        if body not in seen:
            seen.add(body)
            unique.append(note)
    removed = len(self.raw["notes"]) - len(unique)
    self.raw["notes"] = unique
    return removed

from_dict classmethod

from_dict(
    obj: dict[str, Any],
    all_actions: (
        list[DomoDataflow_Action_Base] | None
    ) = None,
) -> DomoDataflow_Action_Base

Create a typed action instance from an API response dict.

When called on the base class directly, dispatches to the registered subclass for the action's 'type' field (polymorphic factory). When called on a concrete subclass, creates that subclass directly.

Subclasses can override _extract_fields() to handle type-specific fields.

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

@classmethod
def from_dict(
    cls,
    obj: dict[str, Any],
    all_actions: list[DomoDataflow_Action_Base] | None = None,
) -> DomoDataflow_Action_Base:
    """Create a typed action instance from an API response dict.

    When called on the base class directly, dispatches to the registered
    subclass for the action's 'type' field (polymorphic factory).
    When called on a concrete subclass, creates that subclass directly.

    Subclasses can override _extract_fields() to handle type-specific fields.
    """
    if cls is DomoDataflow_Action_Base:
        action_type = (
            obj.get("type", "Unknown")
            if isinstance(obj, dict)
            else getattr(obj, "type", "Unknown")
        )
        cls = get_action_class(action_type)
        return cls.from_dict(obj, all_actions=all_actions)

    dd = obj if isinstance(obj, util_dd.DictDot) else util_dd.DictDot(obj)

    # Determine tile_type (category)
    action_type = dd.type
    tile_type = get_action_category(action_type)
    if tile_type is None and cls.__module__:
        # Default to module name if not explicitly registered
        # e.g., 'crew_dcs.classes.DomoDataflow.action.filter' -> 'filter'
        module_parts = cls.__module__.split(".")
        if len(module_parts) > 0:
            tile_type = module_parts[-1]

    # Extract common fields
    instance = cls(
        id=dd.id,
        action_type=action_type,
        tile_type=tile_type,
        name=dd.name or dd.targetTableName or dd.tableName,
        depends_on=dd.dependsOn or [],
        disabled=dd.disabled or False,
        gui=dd.gui,
        settings=dd.settings,
        raw=obj,
    )

    # Let subclasses extract type-specific fields
    instance._extract_fields(dd)

    # Resolve parent actions if provided
    if all_actions:
        instance.get_parents(all_actions)

    return instance

get_parents

get_parents(
    domo_actions: list[DomoDataflow_Action_Base],
) -> list[DomoDataflow_Action_Base] | None

Resolve parent actions from the depends_on IDs.

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

def get_parents(
    self, domo_actions: list[DomoDataflow_Action_Base]
) -> list[DomoDataflow_Action_Base] | None:
    """Resolve parent actions from the depends_on IDs."""
    if self.depends_on and len(self.depends_on) > 0:
        self.parent_actions = [
            parent_action
            for depends_id in self.depends_on
            for parent_action in domo_actions
            if parent_action.id == depends_id
        ]

        if self.parent_actions:
            for parent in self.parent_actions:
                if parent.depends_on:
                    parent.get_parents(domo_actions)

    return self.parent_actions

registered_types classmethod

registered_types() -> list[str]

Return all registered action type strings.

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

@classmethod
def registered_types(cls) -> list[str]:
    """Return all registered action type strings."""
    return sorted(_ACTION_TYPE_REGISTRY.keys())

rewire_upstream

rewire_upstream(upstream_ids: list[str] | str) -> None

Rewire this action's upstream dependencies.

Sets both dependsOn and inputs on the raw dict. Domo normalises dependsOn = inputs on PUT save for PublishToVault actions, so using the upstream action ID (not a table name) in both fields avoids validation errors.

Parameters:

Name	Type	Description	Default
upstream_ids	list[str] | str	A single action ID or list of action IDs to depend on.	required

Example::

action.rewire_upstream(split_join_id)

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

def rewire_upstream(self, upstream_ids: list[str] | str) -> None:
    """Rewire this action's upstream dependencies.

    Sets both ``dependsOn`` and ``inputs`` on the raw dict.  Domo
    normalises ``dependsOn = inputs`` on PUT save for PublishToVault
    actions, so using the upstream *action ID* (not a table name) in
    both fields avoids validation errors.

    Args:
        upstream_ids: A single action ID or list of action IDs to
            depend on.

    Example::

        action.rewire_upstream(split_join_id)
    """
    if self.raw is None:
        raise RuntimeError(
            "Cannot rewire: action has no raw dict (was it created from_dict?)"
        )
    if isinstance(upstream_ids, str):
        upstream_ids = [upstream_ids]
    self.raw["dependsOn"] = list(upstream_ids)
    self.raw["inputs"] = list(upstream_ids)
    self.depends_on = list(upstream_ids)

to_canvas_tile

to_canvas_tile(
    x: int | None = None, y: int | None = None
) -> CanvasTile

Return the typed CanvasTile for this action.

Delegates to CanvasTile.from_action(); pass x/y to override stored position.

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

def to_canvas_tile(self, x: int | None = None, y: int | None = None) -> CanvasTile:
    """Return the typed CanvasTile for this action.

    Delegates to CanvasTile.from_action(); pass x/y to override stored position.
    """
    gui = self.gui or {}
    return CanvasTile.from_action(
        action_id=self.id,
        x=x if x is not None else gui.get("x", 0),
        y=y if y is not None else gui.get("y", 0),
        color=gui.get("color"),
        color_source=gui.get("colorSource"),
    )

unregistered_types classmethod

unregistered_types() -> set[str]

Return action types encountered but not registered (for discovery).

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

@classmethod
def unregistered_types(cls) -> set[str]:
    """Return action types encountered but not registered (for discovery)."""
    return _UNREGISTERED_TYPES.copy()

DomoDataflow_Action_ConcatFields dataclass

DomoDataflow_Action_ConcatFields(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    separator: str = None,
    target_field_name: str = None,
    remove_selected_fields: bool = False,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Concat Fields action - concatenates multiple columns into one.

Also known as "Combine Columns" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	List of columns to concatenate
separator	str	Separator string between values
target_field_name	str	Name of the output column
remove_selected_fields	bool	Whether to remove source columns

Example

concat_action = dataflow.get_action_objects("ConcatFields")[0] print(f"Combining {concat_action.source_columns} with '{concat_action.separator}'")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for ConcatFields (Combine Columns) tile.

Multiple input columns → single output column (TRANSFORMATION).

source_columns property

source_columns: list[str]

Get list of columns being concatenated.

DomoDataflow_Action_Constant dataclass

DomoDataflow_Action_Constant(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Constant action - adds constant value columns.

Also known as "Add Constants" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	List of constant field definitions with name, type, and value

Example

constant_action = dataflow.get_action_objects("Constant")[0] for field in constant_action.fields: ... print(f"{field.get('name')}: {field.get('value')}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for Constant (Add Constants) tile.

Constants produce new columns with no upstream column source. We record them as DIRECT/TRANSFORMATION with to_column=None.

constants property

constants: dict[str, Any]

Get mapping of constant name -> value.

DomoDataflow_Action_DateCalculator dataclass

DomoDataflow_Action_DateCalculator(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    calculations: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Date Calculator action - performs date calculations.

Attributes:

Name	Type	Description
calculations	list[dict]	List of date calculation definitions

Example

date_action = dataflow.get_action_objects("DateCalculator")[0] for calc in date_action.calculations: ... print(f"{calc.get('name')}: {calc.get('operation')}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for DateCalculator tile.

Each calculation produces an output column from input columns.

DomoDataflow_Action_Denormalizer dataclass

DomoDataflow_Action_Denormalizer(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    key_field: str = None,
    group: list[dict] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Denormaliser action - pivots rows to columns.

Also known as "Pivot" in the Magic ETL UI.

Attributes:

Name	Type	Description
key_field	str	Column containing values to become column headers
group	list[dict]	Columns to group by
fields	list[dict]	Columns to pivot

Example

pivot_action = dataflow.get_action_objects("Denormaliser")[0] print(f"Pivoting on '{pivot_action.key_field}'")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for Denormaliser (Pivot) tile.

Group columns → DIRECT/IDENTITY (passthrough) Key field → INDIRECT/GROUP_BY Pivot fields → DIRECT/TRANSFORMATION

group_columns property

group_columns: list[str]

Get list of columns being grouped by.

DomoDataflow_Action_ExpressionEvaluator dataclass

DomoDataflow_Action_ExpressionEvaluator(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    expressions: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Expression Evaluator action - creates calculated columns using formulas.

Also known as "Add Formula" or "Beast Mode" in the Magic ETL UI.

Attributes:

Name	Type	Description
expressions	list[dict]	List of expression definitions with name, expression, and type

Example

formula_action = dataflow.get_action_objects("ExpressionEvaluator")[0] for expr in formula_action.expressions: ... print(f"{expr.get('name')}: {expr.get('expression')}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for ExpressionEvaluator (Add Formula) tile.

Each expression produces a new output column. The expression string may reference multiple input columns. We capture the formula as the expression and mark it as DIRECT/TRANSFORMATION.

formulas property

formulas: dict[str, str]

Get mapping of formula name -> expression.

DomoDataflow_Action_Filter dataclass

DomoDataflow_Action_Filter(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    filter_list: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Filter action - filters rows based on conditions.

Attributes:

Name	Type	Description
filter_list	list[dict]	List of filter conditions

Example

filter_action = dataflow.get_action_objects("Filter")[0] print(filter_action.filter_list)

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for Filter tile.

Filtered columns → INDIRECT/FILTER All other columns → DIRECT/IDENTITY (passthrough)

DomoDataflow_Action_GenerateTable dataclass

DomoDataflow_Action_GenerateTable(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    row_count: int = None,
)

Bases: DomoDataflow_Action_Base

Generate table action for creating data programmatically.

DomoDataflow_Action_GroupBy dataclass

DomoDataflow_Action_GroupBy(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    groups: list[dict] = None,
    fields: list[dict] = None,
    all_rows: bool = False,
    add_line_number: bool = False,
    give_back_row: str = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Group By action - aggregates data by grouping columns.

Attributes:

Name	Type	Description
groups	list[dict]	List of columns to group by
fields	list[dict]	List of aggregation fields with aggregation type
all_rows	bool	Whether to include all rows (not just first per group)

Example

groupby_action = dataflow.get_action_objects("GroupBy")[0] print(f"Grouping by: {groupby_action.group_columns}") print(f"Aggregations: {groupby_action.aggregations}")

aggregations property

aggregations: dict[str, str]

Get mapping of output column -> aggregation type.

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for GroupBy tile.

Group columns → DIRECT/IDENTITY (passthrough) Aggregate fields → DIRECT/AGGREGATION (source column → output column)

group_columns property

group_columns: list[str]

Get list of columns being grouped by.

DomoDataflow_Action_LoadFromVault dataclass

DomoDataflow_Action_LoadFromVault(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    datasource_id: str = None,
    execute_flow_when_updated: bool = False,
    only_load_new_versions: bool = False,
    column_settings: dict = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Input action - loads data from a Domo dataset.

Attributes:

Name	Type	Description
datasource_id	str	The ID of the source dataset
execute_flow_when_updated	bool	Trigger dataflow on dataset update
only_load_new_versions	bool	Only load data when dataset has new data

column_relationships property

column_relationships: set[ColumnRelationship]

IDENTITY relationships from source dataset columns to this tile's output.

LoadFromVault is a passthrough — every column from the source dataset appears in the output unchanged.

DomoDataflow_Action_MLInferenceAction dataclass

DomoDataflow_Action_MLInferenceAction(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    ml_model_id: str = None,
    inference_column: str = None,
    inference_column_rename: str = None,
    inference_response: dict = None,
    include_input_data: bool = True,
    model_schema: dict = None,
    column_settings: dict = None,
    notes: str = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

ML Inference action - runs ML model predictions.

Attributes:

Name	Type	Description
ml_model_id	str	ID of the ML model to use
inference_column	str	Output column name for predictions
include_input_data	bool	Whether to include input columns in output

Example

ml_action = dataflow.get_action_objects("MLInferenceAction")[0] print(f"Model ID: {ml_action.ml_model_id}")

DomoDataflow_Action_MergeJoin dataclass

DomoDataflow_Action_MergeJoin(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    join_type: str = None,
    keys1: list[str] = None,
    keys2: list[str] = None,
    relationship_type: str = None,
    schema_modification1: dict = None,
    schema_modification2: dict = None,
    step1: str = None,
    step2: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Merge Join action - joins two data sources.

Attributes:

Name	Type	Description
join_type	str	Type of join (INNER, LEFT, RIGHT, OUTER)
keys1	list[str]	Join keys from first input
keys2	list[str]	Join keys from second input

Example

join_action = dataflow.get_action_objects("MergeJoin")[0] print(f"Join type: {join_action.join_type}") print(f"Keys: {join_action.keys1} = {join_action.keys2}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for MergeJoin tile.

Join keys → INDIRECT/JOIN (both sides)

DomoDataflow_Action_Metadata dataclass

DomoDataflow_Action_Metadata(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Metadata action - modifies column metadata (rename, type change).

Also known as "Set Column Type" or "Rename Columns" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	List of field metadata modifications

Example

metadata_action = dataflow.get_action_objects("Metadata")[0] for field in metadata_action.fields: ... print(f"{field.get('name')}: {field.get('type')}")

column_relationships property

column_relationships: set[ColumnRelationship]

IDENTITY relationships for Metadata (Alter) tile.

Metadata tiles rename columns and change types. Each field maps from the output column name (rename if present, else original name) to the input column name (original name).

Columns NOT listed in fields are implicitly passed through (Metadata is an "Alter" tile — it only modifies what's explicitly listed).

type_changes property

type_changes: dict[str, str]

Get mapping of column -> new type.

DomoDataflow_Action_NormalizeAll dataclass

DomoDataflow_Action_NormalizeAll(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    id_fields: list[dict] = None,
    key_field: str = None,
    value_field: str = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Normalize All action - unpivots all columns except ID columns.

Also known as "Dynamic Unpivot" in the Magic ETL UI.

Attributes:

Name	Type	Description
id_fields	list[dict]	Columns to keep as identifiers (not unpivoted)
key_field	str	Name of the column for original column names
value_field	str	Name of the column for values

Example

normalize_action = dataflow.get_action_objects("NormalizeAll")[0] print(f"ID columns: {normalize_action.id_columns}")

id_columns property

id_columns: list[str]

Get list of ID columns (not being unpivoted).

DomoDataflow_Action_Normalizer dataclass

DomoDataflow_Action_Normalizer(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    typefield: str = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Normalizer action - unpivots columns to rows.

Also known as "Unpivot" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	Columns to unpivot
typefield	str	Name of the column containing original column names

Example

normalizer_action = dataflow.get_action_objects("Normalizer")[0] print(f"Unpivoting columns into '{normalizer_action.typefield}'")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for Normalizer (Unpivot) tile.

Unpivoted columns → DIRECT/TRANSFORMATION (multiple cols → key+value) typefield column → synthetic (no upstream column)

unpivot_columns property

unpivot_columns: list[str]

Get list of columns being unpivoted.

DomoDataflow_Action_NumericCalculator dataclass

DomoDataflow_Action_NumericCalculator(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    calculations: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Numeric Calculator action - performs numeric calculations.

Also known as "Calculator" in the Magic ETL UI.

Attributes:

Name	Type	Description
calculations	list[dict]	List of calculation definitions

Example

calc_action = dataflow.get_action_objects("NumericCalculator")[0] for calc in calc_action.calculations: ... print(f"{calc.get('name')}: {calc.get('operation')}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for NumericCalculator tile.

Each calculation produces an output column from input columns.

DomoDataflow_Action_PublishToVault dataclass

DomoDataflow_Action_PublishToVault(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    data_source: dict = None,
    version_chain_type: str = None,
    partitioned: bool = False,
    partition_key: str | None = None,
    schema_source: str = None,
    inputs: list[str] = None,
    tables: list[dict] = None,
)

Bases: _PublishBase

Output action - writes data to a Domo dataset.

Attributes:

Name	Type	Description
data_source	dict	Output dataset configuration
version_chain_type	str	How to handle versions (REPLACE, APPEND, etc.)
partitioned	bool	Whether the output is partitioned
partition_key	str | None	Column name used as the partition key (required when partitioned)

DomoDataflow_Action_PublishToWriteback dataclass

DomoDataflow_Action_PublishToWriteback(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    data_source: dict = None,
    version_chain_type: str = None,
    partitioned: bool = False,
    partition_key: str | None = None,
    schema_source: str = None,
    inputs: list[str] = None,
    tables: list[dict] = None,
    writeback_info: dict = None,
)

Bases: _PublishBase

Publish to Writeback action - writes data to external systems.

Similar to PublishToVault but for external destinations (Snowflake, etc.).

Attributes:

Name	Type	Description
writeback_info	dict	Writeback-specific configuration (connection, table, etc.)

DomoDataflow_Action_PythonEngine dataclass

DomoDataflow_Action_PythonEngine(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    script: str = None,
    conda_env: dict = None,
    account_permission: str = None,
    inputs: list[str] = None,
    additions: list[dict] = None,
    remove_by_default: bool = False,
    fill_missing_with_null: bool = False,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Python Engine action - executes Python scripts.

Also known as "Python Script" in the Magic ETL UI.

Attributes:

Name	Type	Description
script	str	Python code to execute
conda_env	dict	Conda environment configuration
inputs	list[str]	List of input action IDs
additions	list[dict]	Output column definitions

Example

python_action = dataflow.get_action_objects("PythonEngineAction")[0] print(python_action.script)

DomoDataflow_Action_ReplaceString dataclass

DomoDataflow_Action_ReplaceString(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Replace String action - replaces strings in columns.

Also known as "Replace Text" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	List of replacement configurations

Example

replace_action = dataflow.get_action_objects("ReplaceString")[0] for field in replace_action.fields: ... print(f"{field.get('name')}: {field.get('find')} -> {field.get('replace')}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for ReplaceString (Replace Text) tile.

Each field is a column where text is replaced in-place.

DomoDataflow_Action_SQL dataclass

DomoDataflow_Action_SQL(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    statements: list[str] = None,
    sql_dialect: str = None,
)

Bases: DomoDataflow_Action_Base

SQL transform action - executes SQL statements.

Attributes:

Name	Type	Description
statements	list[str]	List of SQL statements to execute
sql_dialect	str	SQL dialect (e.g., "MAGIC")

Example

sql_action = dataflow.get_action_objects("SQL")[0] sql_action.sql = "SELECT * FROM Input LIMIT 10" dataflow = await dataflow.update_action(sql_action.name, sql_action.raw)

sql property writable

sql: str | None

Get or set the first SQL statement (convenience property).

Getting returns the first statement or None. Setting updates both self.statements and self.raw.

extract_tables_and_fields

extract_tables_and_fields() -> dict[str, list[str]]

Parse referenced tables and fields from all SQL statements.

Uses sqlglot to extract table and column references. SELECT * yields no explicit field names; * is excluded.

Returns:

Type	Description
dict[str, list[str]]	{"tables": sorted list of table names, "fields": sorted list of column names (empty for SELECT *)}

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

def extract_tables_and_fields(self) -> dict[str, list[str]]:
    """Parse referenced tables and fields from all SQL statements.

    Uses sqlglot to extract table and column references.
    SELECT * yields no explicit field names; * is excluded.

    Returns:
        {"tables": sorted list of table names,
         "fields": sorted list of column names (empty for SELECT *)}
    """
    import sqlglot
    import sqlglot.expressions as sqlexp

    tables: set[str] = set()
    fields: set[str] = set()

    for stmt in self.statements or []:
        try:
            tree = sqlglot.parse_one(stmt, dialect="spark")
        except Exception:  # noqa: BLE001
            continue
        tables.update(t.name for t in tree.find_all(sqlexp.Table) if t.name)
        fields.update(
            c.name for c in tree.find_all(sqlexp.Column) if c.name and c.name != "*"
        )

    return {"tables": sorted(tables), "fields": sorted(fields)}

DomoDataflow_Action_SelectValues dataclass

DomoDataflow_Action_SelectValues(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Select Values action - selects/renames/reorders columns.

Also known as "Select Columns" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	List of field configurations (name, rename, include/exclude)

Example

select_action = dataflow.get_action_objects("SelectValues")[0] for field in select_action.fields: ... print(f"{field.get('name')} -> {field.get('rename', field.get('name'))}")

column_names property

column_names: list[str]

Get list of selected column names.

column_relationships property

column_relationships: set[ColumnRelationship]

IDENTITY relationships for SelectValues (Select Columns) tile.

SelectValues explicitly lists columns to include. Each field maps from the output column name (rename if present, else original name) to the input column name (original name).

Fields with remove=True are excluded from the output.

renamed_columns property

renamed_columns: dict[str, str]

Get mapping of original name -> renamed name for renamed columns.

convert_to_metadata

convert_to_metadata(
    type_overrides: dict[str, str] | None = None,
) -> None

Convert this SelectValues action into a Metadata action in-place.

Metadata (Alter) tiles automatically pass through columns that are not explicitly listed, so only fields that actually rename or change type need to be kept in the explicit list.

Join safety: When the output feeds into a Join or SplitJoin, mismatched types on join keys cause silent wrong results. Use type_overrides to explicitly pin the type of join columns so the assumption either holds or fails hard::

select_action.convert_to_metadata(
    type_overrides={"Page ID": "STRING"}
)

Mutates self.raw["type"] to "Metadata", applies any type_overrides, and filters self.raw["fields"] to only those with a rename or type key. Also updates self.action_type and self.fields on the typed object.

Parameters:

Name	Type	Description	Default
type_overrides	dict[str, str] | None	Mapping of field name → explicit Domo type string (e.g. "STRING", "INT", "DOUBLE"). Applied before the filter so overridden fields are always kept even if they have no rename.	None

Raises:

Type	Description
RuntimeError	If self.raw is None.

Example::

select_action.convert_to_metadata()
select_action.convert_to_metadata(
    type_overrides={"Page ID": "STRING", "User ID": "INT"}
)

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

def convert_to_metadata(
    self,
    type_overrides: dict[str, str] | None = None,
) -> None:
    """Convert this SelectValues action into a Metadata action in-place.

    Metadata (Alter) tiles automatically pass through columns that are
    not explicitly listed, so only fields that actually rename or change
    type need to be kept in the explicit list.

    **Join safety**: When the output feeds into a Join or SplitJoin,
    mismatched types on join keys cause silent wrong results.  Use
    *type_overrides* to explicitly pin the type of join columns so the
    assumption either holds or fails hard::

        select_action.convert_to_metadata(
            type_overrides={"Page ID": "STRING"}
        )

    Mutates ``self.raw["type"]`` to ``"Metadata"``, applies any
    *type_overrides*, and filters ``self.raw["fields"]`` to only those
    with a ``rename`` or ``type`` key.  Also updates ``self.action_type``
    and ``self.fields`` on the typed object.

    Args:
        type_overrides: Mapping of field ``name`` → explicit Domo type
            string (e.g. ``"STRING"``, ``"INT"``, ``"DOUBLE"``).
            Applied **before** the filter so overridden fields are always
            kept even if they have no rename.

    Raises:
        RuntimeError: If ``self.raw`` is None.

    Example::

        select_action.convert_to_metadata()
        select_action.convert_to_metadata(
            type_overrides={"Page ID": "STRING", "User ID": "INT"}
        )
    """
    if self.raw is None:
        raise RuntimeError(
            "Cannot convert: action has no raw dict (was it created from_dict?)"
        )

    # Apply explicit type overrides before filtering
    if type_overrides:
        for f in self.raw.get("fields") or []:
            if f.get("name") in type_overrides:
                f["type"] = type_overrides[f["name"]]

    self.raw["type"] = "Metadata"
    self.raw["fields"] = [
        f
        for f in (self.raw.get("fields") or [])
        if f.get("rename") or f.get("type")
    ]
    self.action_type = "Metadata"
    self.fields = self.raw["fields"]

DomoDataflow_Action_SetValueField dataclass

DomoDataflow_Action_SetValueField(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Set Value Field action - sets or updates field values.

Attributes:

Name	Type	Description
fields	list[dict]	List of field value settings

Example

setvalue_action = dataflow.get_action_objects("SetValueField")[0] for field in setvalue_action.fields: ... print(f"{field.get('name')}: {field.get('value')}")

DomoDataflow_Action_SplitColumn dataclass

DomoDataflow_Action_SplitColumn(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    source_column: str = None,
    delimiter: str = None,
    delimiter_type: str = None,
    use_regex: bool = False,
    additions: list[dict] = None,
    combine_extra_splits: bool = False,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Split Column action - splits a column into multiple columns.

Attributes:

Name	Type	Description
source_column	str	Column to split
delimiter	str	Delimiter to split on
delimiter_type	str	Type of delimiter (STRING, REGEX, etc.)
use_regex	bool	Whether delimiter is a regex
additions	list[dict]	New columns to create from splits
combine_extra_splits	bool	Whether to combine extra splits into last column

Example

split_action = dataflow.get_action_objects("SplitColumnAction")[0] print(f"Splitting {split_action.source_column} on '{split_action.delimiter}'")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for SplitColumn tile.

One input column → multiple output columns (TRANSFORMATION).

output_columns property

output_columns: list[str]

Get list of output column names.

DomoDataflow_Action_SplitJoin dataclass

DomoDataflow_Action_SplitJoin(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    keys1: list[str] = None,
    keys2: list[str] = None,
    step1: str = None,
    step2: str = None,
    inner_table: str = None,
    left_anti_table: str = None,
    right_anti_table: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Split Join action - performs a join that splits into multiple outputs.

Creates inner, left anti, and right anti outputs from a single join.

Attributes:

Name	Type	Description
keys1	list[str]	Join keys from first input
keys2	list[str]	Join keys from second input
inner_table	str	Name of inner join output
left_anti_table	str	Name of left anti join output
right_anti_table	str	Name of right anti join output

Example

splitjoin_action = dataflow.get_action_objects("SplitJoin")[0] print(f"Inner: {splitjoin_action.inner_table}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for SplitJoin tile.

Join keys → INDIRECT/JOIN (both sides) Non-key columns from both inputs → DIRECT/IDENTITY (passthrough)

SplitJoin produces three outputs (inner, leftAnti, rightAnti). The join keys are INDIRECT because they're used in the join condition, not as output values.

DomoDataflow_Action_TextFormatting dataclass

DomoDataflow_Action_TextFormatting(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Text Formatting action - applies text formatting transformations.

Attributes:

Name	Type	Description
fields	list[dict]	List of text formatting configurations

Example

text_action = dataflow.get_action_objects("TextFormatting")[0] for field in text_action.fields: ... print(f"{field.get('name')}: {field.get('operation')}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for TextFormatting tile.

Each field is a column where text formatting is applied in-place.

DomoDataflow_Action_TextGeneration dataclass

DomoDataflow_Action_TextGeneration(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    model_id: str = None,
    prompt: str = None,
    instructions: str = None,
    temperature: float = None,
    output_column_name: str = None,
    parameters: dict = None,
    path: str = None,
    model_input_schema: dict = None,
    model_output_schema: dict = None,
    column_settings: dict = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Text Generation action - generates text using AI models.

Attributes:

Name	Type	Description
model_id	str	ID of the AI model
prompt	str	Prompt template for generation
instructions	str	System instructions
temperature	float	Sampling temperature
output_column_name	str	Name of output column

Example

textgen_action = dataflow.get_action_objects("TextGeneration")[0] print(f"Model: {textgen_action.model_id}") print(f"Prompt: {textgen_action.prompt}")

DomoDataflow_Action_UnionAll dataclass

DomoDataflow_Action_UnionAll(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    inputs: list[str] = None,
    union_type: str = None,
    strict: bool = False,
    schema_source: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Union All action - combines multiple data sources vertically.

Attributes:

Name	Type	Description
inputs	list[str]	List of input action IDs
union_type	str	Type of union (UNION_ALL, UNION, etc.)
strict	bool	Whether to enforce strict schema matching

Example

union_action = dataflow.get_action_objects("UnionAll")[0] print(f"Combining {len(union_action.inputs)} inputs")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for UnionAll tile.

UnionAll stacks rows from multiple inputs. Each output column maps to the same-named column in each input (IDENTITY).

DomoDataflow_Action_Unique dataclass

DomoDataflow_Action_Unique(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    fields: list[dict] = None,
    count_rows: bool = False,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Unique action - removes duplicate rows.

Also known as "Remove Duplicates" in the Magic ETL UI.

Attributes:

Name	Type	Description
fields	list[dict]	Columns to consider for uniqueness
count_rows	bool	Whether to add a count column

Example

unique_action = dataflow.get_action_objects("Unique")[0] print(f"Deduping on: {unique_action.dedup_columns}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for Unique (Remove Duplicates) tile.

Dedup columns → INDIRECT/GROUP_BY (used for deduplication)

dedup_columns property

dedup_columns: list[str]

Get list of columns used for deduplication.

DomoDataflow_Action_Unknown dataclass

DomoDataflow_Action_Unknown(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
)

Bases: DomoDataflow_Action_Base

Fallback action class for unregistered action types.

This class is used when an action type is encountered that hasn't been registered. All fields from the API response are preserved in the 'raw' attribute.

DomoDataflow_Action_UserDefined dataclass

DomoDataflow_Action_UserDefined(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    action_definition_id: str = None,
    variables: dict = None,
    inputs: list[str] = None,
    additions: list[dict] = None,
    remove_by_default: bool = False,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

User Defined Action - custom reusable action from Action Library.

Attributes:

Name	Type	Description
action_definition_id	str	ID of the action definition
variables	dict	Variable values for the action
inputs	list[str]	List of input action IDs
additions	list[dict]	Output column definitions

Example

uda_action = dataflow.get_action_objects("UserDefinedAction")[0] print(f"Action Definition: {uda_action.action_definition_id}")

DomoDataflow_Action_ValueMapper dataclass

DomoDataflow_Action_ValueMapper(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    field_to_use: str = None,
    target_field: str = None,
    target_type: str = None,
    mappings: list[dict] = None,
    unmapped_behavior: str = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Value Mapper action - maps values from one set to another.

Also known as "Map Values" in the Magic ETL UI.

Attributes:

Name	Type	Description
field_to_use	str	Source column to map from
target_field	str	Target column name
mappings	list[dict]	List of value mappings
unmapped_behavior	str	What to do with unmapped values

Example

mapper_action = dataflow.get_action_objects("ValueMapper")[0] print(f"Mapping {mapper_action.field_to_use} -> {mapper_action.target_field}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for ValueMapper (Map Values) tile.

Maps values from source column to target column.

value_map property

value_map: dict[Any, Any]

Get mapping of source value -> target value.

DomoDataflow_Action_WindowAction dataclass

DomoDataflow_Action_WindowAction(
    id: str,
    action_type: str = None,
    tile_type: str | None = None,
    name: str = None,
    depends_on: list[str] = None,
    disabled: bool = False,
    gui: dict = None,
    settings: dict = None,
    raw: dict = None,
    parent_actions: list[DomoDataflow_Action_Base] = None,
    additions: list[dict] = None,
    group_rules: list[dict] = None,
    order_rules: list[dict] = None,
    input: str = None,
    tables: list[dict] = None,
)

Bases: DomoDataflow_Action_Base

Window action - performs window/ranking functions.

Also known as "Rank & Window" in the Magic ETL UI.

Attributes:

Name	Type	Description
additions	list[dict]	List of window function definitions
group_rules	list[dict]	Columns to partition by
order_rules	list[dict]	Columns to order by within partitions

Example

window_action = dataflow.get_action_objects("WindowAction")[0] print(f"Partitioning by: {window_action.partition_columns}")

column_relationships property

column_relationships: set[ColumnRelationship]

Column relationships for WindowAction (Rank & Window) tile.

Partition columns → INDIRECT/WINDOW Order columns → INDIRECT/SORT Addition (window function) columns → DIRECT/TRANSFORMATION

order_columns property

order_columns: list[str]

Get list of columns used for ordering.

partition_columns property

partition_columns: list[str]

Get list of columns used for partitioning.

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

register_action_type

register_action_type(
    action_type: str, category: str | None = None
)

Decorator to register a DomoDataflow_Action_Base subclass.

Parameters:

Name	Type	Description	Default
action_type	str	The action type identifier (e.g., 'LoadFromVault', 'Filter')	required
category	str | None	The tile category/type (e.g., 'filter', 'aggregate', 'pivot'). Defaults to the module folder name where the action is defined.	None

Example

@register_action_type('Filter', category='filter') @dataclass class DomoDataflow_Action_Filter(DomoDataflow_Action_Base): filter_list: list[dict] = None

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

def register_action_type(action_type: str, category: str | None = None):
    """Decorator to register a DomoDataflow_Action_Base subclass.

    Args:
        action_type: The action type identifier (e.g., 'LoadFromVault', 'Filter')
        category: The tile category/type (e.g., 'filter', 'aggregate', 'pivot').
                  Defaults to the module folder name where the action is defined.

    Example:
        @register_action_type('Filter', category='filter')
        @dataclass
        class DomoDataflow_Action_Filter(DomoDataflow_Action_Base):
            filter_list: list[dict] = None
    """

    def decorator(
        cls: type[DomoDataflow_Action_Base],
    ) -> type[DomoDataflow_Action_Base]:
        _ACTION_TYPE_REGISTRY[action_type] = cls

        # Store category if provided, otherwise use None (will default to module name)
        if category:
            _ACTION_CATEGORY_REGISTRY[action_type] = category

        return cls

    return decorator

Modules

• aggregate
• ai_services
• base
• combine
• data_science
• datasets
• dates_numbers
• filter
• manager
• pivot
• result
• scripting
• sql_converter
• text
• utility