view_definition

view_definition

Typed classes for Domo dataset view definitions.

Models the /api/query/v1/datasources/{id}/schema/indexed API response as structured dataclasses with from_dict classmethods, replacing raw dict access with typed properties and methods.

Structure

ViewDefinition ├── columns: list[ViewColumn] # view's output columns ├── from_item: ViewFromItem # base dataset + alias ├── joins: list[ViewJoin] # joined datasets ├── select_string: str # SQL SELECT statement ├── column_mapping: dict[str, ViewColumnMapping] # view col → source │ └── ViewColumnMapping │ ├── formatted_expression: str # e.g., "base.Name" │ ├── source_dataset_id: str # resolved from alias │ └── source_column: str # resolved column name ├── where_columns: dict[str, ViewColumnMapping] ├── having_columns: dict[str, ViewColumnMapping] ├── calculated_columns: dict[str, ViewColumnMapping] └── remapped_columns: dict[str, ViewColumnMapping]

ViewColumn (from tables[0].columns) ├── name, id, type ├── visible, aggregated, order └── reference_data_source_id

ViewFromItem ├── dataset_id: str └── alias: str (e.g., "base")

ViewJoin ├── dataset_id: str ├── join_type: str (e.g., "left") ├── left_column: str └── right_column: str

ViewColumn dataclass

ViewColumn(
    name: str = "",
    id: str = "",
    type: str | None = None,
    visible: bool = True,
    aggregated: bool = False,
    order: int = 0,
    reference_data_source_id: str | None = None,
)

A column in the view's output schema (from tables[0].columns).

Attributes:

Name	Type	Description
name	str	Column display name
id	str	Column ID (usually same as name)
type	str | None	Data type (e.g., "STRING", "DOUBLE", "DATE")
visible	bool	Whether the column is visible
aggregated	bool	Whether the column is aggregated
order	int	Column order in the view
reference_data_source_id	str | None	The view's own dataset ID

ViewColumnMapping dataclass

ViewColumnMapping(
    formatted_expression: str = "",
    data_type: str | None = None,
    aggregated: bool = False,
    source_dataset_id: str | None = None,
    source_column: str | None = None,
)

Maps a view column to its source dataset and column.

Parsed from viewTemplate.fromItemInfo.mapping.columnInfo.

Attributes:

Name	Type	Description
formatted_expression	str	Raw expression (e.g., `base`.`Name`)
data_type	str | None	Data type of the column (e.g., "STRING", "DOUBLE")
aggregated	bool	Whether the column is aggregated
source_dataset_id	str | None	Resolved dataset ID (from alias lookup)
source_column	str | None	Resolved source column name (backtick-stripped)

from_dict classmethod

from_dict(
    obj: dict[str, Any],
    alias_to_dataset_id: dict[str, str] | None = None,
) -> ViewColumnMapping

Build a ViewColumnMapping from a columnInfo entry.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	The columnInfo dict for one column (e.g., {"formattedExpression": "base.Name", "type": "STRING"})	required
alias_to_dataset_id	dict[str, str] | None	Mapping of alias → dataset ID (e.g., {"base": "8e9a51c3-..."})	None

Source code in src/crew_dcs/classes/DomoDataset/view_definition.py

@classmethod
def from_dict(
    cls,
    obj: dict[str, Any],
    alias_to_dataset_id: dict[str, str] | None = None,
) -> ViewColumnMapping:
    """Build a ViewColumnMapping from a columnInfo entry.

    Args:
        obj: The columnInfo dict for one column (e.g., {"formattedExpression": "`base`.`Name`", "type": "STRING"})
        alias_to_dataset_id: Mapping of alias → dataset ID (e.g., {"base": "8e9a51c3-..."})
    """
    expr = obj.get("formattedExpression", "")
    alias_or_id, col_name = _parse_formatted_expression(expr)

    # Resolve alias to dataset ID
    source_dataset_id = None
    if alias_or_id:
        if alias_to_dataset_id and alias_or_id in alias_to_dataset_id:
            source_dataset_id = alias_to_dataset_id[alias_or_id]
        else:
            # If not found in alias map, the alias might be a raw dataset ID
            source_dataset_id = alias_or_id

    return cls(
        formatted_expression=expr,
        data_type=obj.get("type"),
        aggregated=obj.get("aggregated", False),
        source_dataset_id=source_dataset_id,
        source_column=col_name,
    )

ViewDefinition dataclass

ViewDefinition(
    name: str = "",
    data_source_id: str = "",
    version_id: str = "",
    columns: list[ViewColumn] = list(),
    from_item: ViewFromItem = ViewFromItem(),
    joins: list[ViewJoin] = list(),
    select_string: str = "",
    column_mapping: dict[str, ViewColumnMapping] = dict(),
    where_columns: dict[str, ViewColumnMapping] = dict(),
    having_columns: dict[str, ViewColumnMapping] = dict(),
    calculated_columns: dict[
        str, ViewColumnMapping
    ] = dict(),
    remapped_columns: dict[str, ViewColumnMapping] = dict(),
)

Typed representation of a dataset view's definition.

Provides structured access to the view's columns, source datasets, and column mappings with convenience properties for lineage queries.

Usage

vd = ViewDefinition.from_dict(api_response) vd.used_columns_by_dataset # dict[str, set[str]] vd.used_columns # set[str] vd.source_dataset_ids # set[str] vd.column_mapping['Name'] # ViewColumnMapping

available_columns property

available_columns: list[str]

All available column names in the view's output schema.

column_relationships property

column_relationships: set[ColumnRelationship]

Set of ColumnRelationship edges for this view.

Direction: from = this view (downstream consumer), to = source dataset (upstream).

Produces: - DIRECT/IDENTITY for each column_mapping entry (SELECT passthrough) - DIRECT/TRANSFORMATION for calculated columns (beast modes) - INDIRECT/FILTER for where_columns - INDIRECT/CONDITIONAL for having_columns - INDIRECT/JOIN for join ON columns

source_dataset_ids property

source_dataset_ids: set[str]

Set of all source dataset IDs this view depends on.

used_columns property

used_columns: set[str]

Flat set of all source column names used across all datasets.

used_columns_by_dataset property

used_columns_by_dataset: dict[str, set[str]]

Map of dataset_id -> set of source column names used from that dataset.

This is the primary lineage property: it tells you which columns from which source datasets the view depends on.

Includes: - All columns from column_mapping (SELECT) - All columns from where_columns (WHERE) - All columns from having_columns (HAVING) - All columns from calculated_columns (beast modes) - Join ON columns (from both sides)

from_dict classmethod

from_dict(obj: dict[str, Any]) -> ViewDefinition

Build a ViewDefinition from the schema/indexed API response.

Parameters:

Name	Type	Description	Default
obj	dict[str, Any]	Full API response dict from /api/query/v1/datasources/{id}/schema/indexed	required

Source code in src/crew_dcs/classes/DomoDataset/view_definition.py

@classmethod
def from_dict(cls, obj: dict[str, Any]) -> ViewDefinition:
    """Build a ViewDefinition from the schema/indexed API response.

    Args:
        obj: Full API response dict from /api/query/v1/datasources/{id}/schema/indexed
    """
    # Parse columns from tables[0].columns
    tables = obj.get("tables", [])
    columns = []
    if tables and isinstance(tables[0], dict):
        columns = [ViewColumn.from_dict(c) for c in tables[0].get("columns", [])]

    # Parse viewTemplate
    vt = obj.get("viewTemplate", {})
    select_body = vt.get("select", {}).get("selectBody", {})

    # fromItem
    from_item = ViewFromItem.from_dict(select_body.get("fromItem", {}))

    # Build alias-to-dataset-id mapping
    # fromItem gives us: alias -> dataset_id
    alias_to_dataset_id: dict[str, str] = {}
    if from_item.alias and from_item.dataset_id:
        alias_to_dataset_id[from_item.alias] = from_item.dataset_id

    # joins give us: dataset_id (no alias in viewTemplate joins)
    joins = [ViewJoin.from_dict(j) for j in select_body.get("joins", [])]
    for join in joins:
        if join.dataset_id:
            # Joins reference datasets by ID directly (no alias)
            # but the formattedExpression might use the ID as an alias
            alias_to_dataset_id[join.dataset_id] = join.dataset_id

    # selectString
    select_string = vt.get("selectString", "")

    # Parse column mappings from fromItemInfo.mapping
    from_item_info = vt.get("fromItemInfo", {})
    mapping = from_item_info.get("mapping", {})

    def _parse_column_info(
        info: dict[str, Any],
    ) -> dict[str, ViewColumnMapping]:
        col_info = info.get("columnInfo", {})
        return {
            name: ViewColumnMapping.from_dict(
                m, alias_to_dataset_id=alias_to_dataset_id
            )
            for name, m in col_info.items()
            if isinstance(m, dict)
        }

    column_mapping = _parse_column_info(mapping)
    where_columns = _parse_column_info(mapping.get("where", {}))
    having_columns = _parse_column_info(mapping.get("having", {}))
    calculated_columns = _parse_column_info(mapping.get("calculated", {}))
    remapped_columns = _parse_column_info(mapping.get("remapped", {}))

    return cls(
        name=obj.get("name", ""),
        data_source_id=obj.get("dataSourceId", ""),
        version_id=str(obj.get("versionId", "")),
        columns=columns,
        from_item=from_item,
        joins=joins,
        select_string=select_string,
        column_mapping=column_mapping,
        where_columns=where_columns,
        having_columns=having_columns,
        calculated_columns=calculated_columns,
        remapped_columns=remapped_columns,
    )

ViewFromItem dataclass

ViewFromItem(
    dataset_id: str = "", alias: str | None = None
)

The base dataset in a view's FROM clause.

Attributes:

Name	Type	Description
dataset_id	str	The dataset ID (backtick-stripped)
alias	str | None	The alias used in SQL (e.g., "base")

ViewJoin dataclass

ViewJoin(
    dataset_id: str = "",
    join_type: str | None = None,
    left_column: str | None = None,
    right_column: str | None = None,
)

A JOIN in the view's query.

Attributes:

Name	Type	Description
dataset_id	str	The joined dataset ID (backtick-stripped)
join_type	str | None	Join type (e.g., "left", "inner")
left_column	str | None	Column name from the left/base side of the ON expression
right_column	str | None	Column name from the right/joined side of the ON expression

build_view_put_schema

build_view_put_schema(
    get_response: dict[str, Any],
    table_aliases: dict[str, str],
    last_modified: str = "",
) -> dict[str, Any]

Normalize a schema/indexed GET response into a PUT-ready schema.

The read endpoint is internally inconsistent for re-PUT; this reconciles it:

1. Strip backticks from selectItems column names (aggregated columns come back backticked, but tables uses plain names -> "column names do not match").
2. Reorder tables columns to match the selectItems order.
3. Add type to each join (left flag -> "left" / "inner").
4. Annotate each selectItem expression with type: {dataType} from the column mapping.
5. Ensure fromItemInfo.mapping carries the empty where/having/calculated/remapped blocks the PUT expects.

Parameters:

Name	Type	Description	Default
get_response	dict[str, Any]	A get_dataset_view_schema_indexed response.	required
table_aliases	dict[str, str]	Map of source-dataset display name -> dataset id.	required
last_modified	str	Optional lastModified echo value.	''

Returns:

Type	Description
dict[str, Any]	A schema dict suitable for update_view.

Source code in src/crew_dcs/classes/DomoDataset/view_definition.py

def build_view_put_schema(
    get_response: dict[str, Any],
    table_aliases: dict[str, str],
    last_modified: str = "",
) -> dict[str, Any]:
    """Normalize a ``schema/indexed`` GET response into a PUT-ready ``schema``.

    The read endpoint is internally inconsistent for re-PUT; this reconciles it:

    1. Strip backticks from ``selectItems`` column names (aggregated columns come
       back backticked, but ``tables`` uses plain names -> "column names do not
       match").
    2. Reorder ``tables`` columns to match the ``selectItems`` order.
    3. Add ``type`` to each join (``left`` flag -> ``"left"`` / ``"inner"``).
    4. Annotate each selectItem ``expression`` with ``type: {dataType}`` from the
       column mapping.
    5. Ensure ``fromItemInfo.mapping`` carries the empty
       ``where``/``having``/``calculated``/``remapped`` blocks the PUT expects.

    Args:
        get_response: A ``get_dataset_view_schema_indexed`` response.
        table_aliases: Map of source-dataset display name -> dataset id.
        last_modified: Optional ``lastModified`` echo value.

    Returns:
        A ``schema`` dict suitable for ``update_view``.
    """
    vt = copy.deepcopy(get_response["viewTemplate"])
    select_body = vt["select"]["selectBody"]

    mapping = vt.setdefault("fromItemInfo", {}).setdefault("mapping", {})
    mapping.setdefault("columnInfo", {})
    col_types = {k: v.get("type") for k, v in mapping["columnInfo"].items()}
    for block in ("where", "having", "calculated", "remapped"):
        mapping.setdefault(block, {"columnInfo": {}})

    for join in select_body.get("joins", []):
        join.setdefault("type", "left" if join.get("left") else "inner")

    select_order: list[str] = []
    for item in select_body.get("selectItems", []):
        expr = item.get("expression", {})
        col = _strip_backticks(expr.get("columnName"))
        expr["columnName"] = col
        select_order.append(col)
        if col in col_types and "type" not in expr:
            expr["type"] = {"dataType": col_types[col]}

    # Reorder tables columns to match the (debacktick'd) select order.
    src_tables = copy.deepcopy(get_response.get("tables") or [{}])
    first = src_tables[0] if src_tables else {}
    by_name = {c.get("name"): c for c in first.get("columns", [])}
    ordered = [by_name[name] for name in select_order if name in by_name]
    # keep any columns not referenced in select (defensive) at the end
    ordered += [
        c for c in first.get("columns", []) if c.get("name") not in select_order
    ]
    for col in ordered:
        col.setdefault("isDataControlled", False)
    tables = [{**first, "columns": ordered}]

    return {
        "dataSourceId": get_response.get("dataSourceId", ""),
        "lastModified": last_modified,
        "tables": tables,
        "viewTemplate": vt,
        "tableAliases": dict(table_aliases),
    }