sql_converter

sql_converter

SQL conversion helpers for DomoDataflow actions.

These helpers provide a lightweight approximation layer that maps Magic ETL action tiles to SQL snippets. The output is intended for analysis and documentation, not direct execution without review.

convert_entire_workflow_to_sql_tiles

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

Convert a full action list into ordered SQL tile approximations.

Each result dict contains

step_number, step_name, tile_id, tile_name, tile_type, input_steps, input_names, output_steps, output_names, conversion_status, sql

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

def convert_entire_workflow_to_sql_tiles(actions: list[Any]) -> list[dict[str, Any]]:
    """Convert a full action list into ordered SQL tile approximations.

    Each result dict contains:
        step_number, step_name, tile_id, tile_name, tile_type,
        input_steps, input_names, output_steps, output_names,
        conversion_status, sql
    """
    ordered = _topo_sort_actions(actions)
    step_lookup = {
        _tile_id(action): f"step_{idx}" for idx, action in enumerate(ordered, start=1)
    }
    name_lookup = {_tile_id(action): _tile_name(action) for action in ordered}

    # Build reverse edge map: tile_id → list of step_names that consume it
    _children: dict[str, list[str]] = defaultdict(list)
    _child_names: dict[str, list[str]] = defaultdict(list)
    for action in ordered:
        aid = _tile_id(action)
        raw = _as_raw_dict(action)
        depends_on = getattr(action, "depends_on", None)
        if depends_on is None:
            depends_on = raw.get("dependsOn", [])
        for pid in depends_on or []:
            child_step = step_lookup.get(aid, aid)
            child_name = name_lookup.get(aid, aid)
            parent_id = str(pid)
            if child_step not in _children[parent_id]:
                _children[parent_id].append(child_step)
            if child_name not in _child_names[parent_id]:
                _child_names[parent_id].append(child_name)

    result: list[dict[str, Any]] = []
    for idx, action in enumerate(ordered, start=1):
        aid = _tile_id(action)
        raw = _as_raw_dict(action)
        depends_on = getattr(action, "depends_on", None)
        if depends_on is None:
            depends_on = raw.get("dependsOn", [])

        input_steps = [
            step_lookup.get(str(pid), str(pid)) for pid in (depends_on or [])
        ] or ["None"]
        input_names = [
            name_lookup.get(str(pid), str(pid)) for pid in (depends_on or [])
        ] or ["None"]

        output_steps = _children.get(aid) or ["Final / Output"]
        output_names = _child_names.get(aid) or ["Final / Output"]

        previous_step = "source_dataset"
        if input_steps and input_steps[0] != "None":
            previous_step = input_steps[0]

        sql = tile_to_sql(
            action,
            previous_step=previous_step,
            input_steps=input_steps,
            input_names=input_names,
        )

        result.append(
            {
                "step_number": idx,
                "step_name": f"step_{idx}",
                "tile_id": aid,
                "tile_name": _tile_name(action),
                "tile_type": _tile_type(action),
                "input_steps": input_steps,
                "input_names": input_names,
                "output_steps": output_steps,
                "output_names": output_names,
                "conversion_status": conversion_status(action),
                "sql": sql,
            }
        )

    return result

tile_to_sql

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

Convert a single tile/action into SQL approximation.

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

def tile_to_sql(
    action_or_dict: Any,
    *,
    previous_step: str = "source_dataset",
    input_steps: list[str] | None = None,
    input_names: list[str] | None = None,
) -> str:
    """Convert a single tile/action into SQL approximation."""
    tile = _as_raw_dict(action_or_dict)
    name = _tile_name(action_or_dict)
    tile_type = _tile_type(action_or_dict).lower()
    input_steps = input_steps or [previous_step]
    input_names = input_names or ["None"]
    previous_step = (
        input_steps[0] if input_steps and input_steps[0] != "None" else previous_step
    )

    if tile_type == "loadfromvault":
        return _loadfromvault_to_sql(tile, name)
    if tile_type in ("metadata", "altercolumns", "setcolumntype", "cast"):
        return _metadata_to_sql(tile, name, previous_step)
    if tile_type in ("groupby", "aggregate"):
        return _groupby_to_sql(tile, name, previous_step)
    if tile_type in ("publishtovault", "outputdataset", "datasetoutput", "output"):
        return _publishtovault_to_sql(tile, name, previous_step)
    if tile_type in ("mergejoin", "join", "joindata"):
        return _mergejoin_to_sql(tile, name, input_steps)

    # Reuse conservative fallback behavior for unsupported/unknown tile types.
    return f"""-- {name}
-- Unsupported or unknown Magic ETL tile type: {_tile_type(action_or_dict)}
-- This step requires manual review.
SELECT *
FROM {previous_step};"""