base

base

DomoDataflow Action Base Classes and Registry

This module provides the registration pattern for Magic ETL v2 action types. Each action type can be registered via the @register_action_type decorator, allowing for graceful handling of unknown action types while providing typed access to known ones.

Usage

Get the appropriate action class for a type

action_class = get_action_class("LoadFromVault") action = action_class.from_dict(raw_action_dict)

Or use the factory function

action = create_action_from_dict(raw_action_dict)

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_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.

create_action_from_dict

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

Factory function to create the appropriate action instance from a dict.

This is the recommended way to create action instances from API responses. It automatically selects the appropriate class based on the 'type' field.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	Raw action dict from the API	required
all_actions	list[DomoDataflow_Action_Base] | None	Optional list of already-created actions for dependency resolution	None

Returns:

Type	Description
DomoDataflow_Action_Base	Appropriate DomoDataflow_Action_* instance

Example

for raw_action in dataflow.raw['procedures'][0]['actions']: ... action = create_action_from_dict(raw_action) ... print(f"{action.action_type}: {action.name}")

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

def create_action_from_dict(
    obj: dict[str, Any], all_actions: list[DomoDataflow_Action_Base] | None = None
) -> DomoDataflow_Action_Base:
    """Factory function to create the appropriate action instance from a dict.

    This is the recommended way to create action instances from API responses.
    It automatically selects the appropriate class based on the 'type' field.

    Args:
        obj: Raw action dict from the API
        all_actions: Optional list of already-created actions for dependency resolution

    Returns:
        Appropriate DomoDataflow_Action_* instance

    Example:
        >>> for raw_action in dataflow.raw['procedures'][0]['actions']:
        ...     action = create_action_from_dict(raw_action)
        ...     print(f"{action.action_type}: {action.name}")
    """
    action_type = obj.get("type", "Unknown")
    cls = get_action_class(action_type)
    return cls.from_dict(obj, all_actions=all_actions)

get_action_category

get_action_category(action_type: str) -> str | None

Get the registered category (TileType) for a given action type.

Parameters:

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

Returns:

Type	Description
str | None	The category string (e.g., 'filter', 'aggregate') or None if not registered

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

def get_action_category(action_type: str) -> str | None:
    """Get the registered category (TileType) for a given action type.

    Args:
        action_type: The action type string (e.g., "LoadFromVault")

    Returns:
        The category string (e.g., 'filter', 'aggregate') or None if not registered
    """
    return _ACTION_CATEGORY_REGISTRY.get(action_type)

get_action_class

get_action_class(
    action_type: str,
) -> type[DomoDataflow_Action_Base]

Get the registered action class for a given type.

If the type is not registered, returns DomoDataflow_Action_Unknown and tracks it for discovery purposes.

Parameters:

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

Returns:

Type	Description
type[DomoDataflow_Action_Base]	The registered action class, or DomoDataflow_Action_Unknown

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

def get_action_class(action_type: str) -> type[DomoDataflow_Action_Base]:
    """Get the registered action class for a given type.

    If the type is not registered, returns DomoDataflow_Action_Unknown
    and tracks it for discovery purposes.

    Args:
        action_type: The action type string (e.g., "LoadFromVault")

    Returns:
        The registered action class, or DomoDataflow_Action_Unknown
    """
    cls = _ACTION_TYPE_REGISTRY.get(action_type)
    if cls is None:
        _UNREGISTERED_TYPES.add(action_type)
        # Note: Using warnings.warn() because this is a sync function.
        # cl is async and cannot be used in sync contexts.
        return DomoDataflow_Action_Unknown
    return cls

get_registered_action_types

get_registered_action_types() -> list[str]

Get list of all registered action types.

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

def get_registered_action_types() -> list[str]:
    """Get list of all registered action types."""
    return sorted(_ACTION_TYPE_REGISTRY.keys())

get_unregistered_action_types

get_unregistered_action_types() -> set[str]

Get set of action types that were encountered but not registered.

Useful for discovering new action types that need to be implemented.

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

def get_unregistered_action_types() -> set[str]:
    """Get set of action types that were encountered but not registered.

    Useful for discovering new action types that need to be implemented.
    """
    return _UNREGISTERED_TYPES.copy()

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