column_lineage

column_lineage

Column-level lineage mixin for DomoLineage.

Provides get_column_lineage() and to_erd() methods for tracing column-level relationships through view definitions and dataflow actions.

Extracted from base.py for separation of concerns.

DomoLineage_ColumnMixin

Column-level lineage methods for DomoLineage.

Provides methods for tracing column relationships through view definitions and dataflow actions, and generating Mermaid ER diagrams from those relationships.

This mixin is mixed into DomoLineage in base.py.

get_column_lineage async

get_column_lineage(
    *,
    column_name: str | None = None,
    include_view_definitions: bool = True,
    include_dataflows: bool = True,
    trace_upstream: bool = True,
    session: AsyncClient = None,
    debug_api: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
) -> set[ColumnRelationship]

Get column-level lineage for this entity.

Traverses the entity-level lineage and extracts column relationships from view definitions and dataflow actions.

Parameters:

Name	Type	Description	Default
column_name	str | None	Optional column name to filter/trace. If provided, returns only relationships involving this column and traces upstream from it.	None
include_view_definitions	bool	Include column relationships from dataset view definitions (beast modes, calculated fields).	True
include_dataflows	bool	Include column relationships from dataflow actions (ETL transformations).	True
trace_upstream	bool	If True and column_name is provided, trace only the upstream chain for that column.	True
session	AsyncClient	HTTP session for reuse.	None
debug_api	bool	Enable API debug logging.	False
context	RouteContext | None	Route context for API configuration.	None

Returns:

Type	Description
set[ColumnRelationship]	Set of ColumnRelationship objects representing column-level lineage.

Example

Get all column relationships for a card's dataset

rels = await card.dataset.Lineage.get_column_lineage()

Trace a specific column upstream

rels = await card.dataset.Lineage.get_column_lineage( ... column_name="Revenue", ... trace_upstream=True, ... )

Source code in src/crew_dcs/classes/subentity/lineage/column_lineage.py

async def get_column_lineage(  # noqa: C901
    self,
    *,
    column_name: str | None = None,
    include_view_definitions: bool = True,
    include_dataflows: bool = True,
    trace_upstream: bool = True,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
) -> set[ColumnRelationship]:
    """Get column-level lineage for this entity.

    Traverses the entity-level lineage and extracts column relationships
    from view definitions and dataflow actions.

    Args:
        column_name: Optional column name to filter/trace. If provided,
                    returns only relationships involving this column
                    and traces upstream from it.
        include_view_definitions: Include column relationships from
                                   dataset view definitions (beast modes,
                                   calculated fields).
        include_dataflows: Include column relationships from dataflow
                          actions (ETL transformations).
        trace_upstream: If True and column_name is provided, trace only
                       the upstream chain for that column.
        session: HTTP session for reuse.
        debug_api: Enable API debug logging.
        context: Route context for API configuration.

    Returns:
        Set of ColumnRelationship objects representing column-level lineage.

    Example:
        >>> # Get all column relationships for a card's dataset
        >>> rels = await card.dataset.Lineage.get_column_lineage()
        >>> # Trace a specific column upstream
        >>> rels = await card.dataset.Lineage.get_column_lineage(
        ...     column_name="Revenue",
        ...     trace_upstream=True,
        ... )
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    # Ensure entity-level lineage is loaded
    if not self.lineage:
        await self.get(
            session=session,
            debug_api=debug_api,
            context=context,
        )

    all_rels: set[ColumnRelationship] = set()

    # Extract column relationships from each entity in lineage
    for link in self.lineage:
        entity = link.entity
        if entity is None:
            continue

        # View definitions (dataset views, card datasets)
        if include_view_definitions and hasattr(entity, "ViewDefinition"):
            try:
                view_def = await entity.ViewDefinition.get(
                    context=context,
                    debug_api=debug_api,
                )
                if view_def:
                    all_rels.update(view_def.column_relationships)
            except Exception as e:  # noqa: BLE001
                await logger.warning(
                    f"Failed to get view definition for {link.type} {link.id}: {e}"
                )

        # Dataflow actions
        if include_dataflows and hasattr(entity, "Actions"):
            try:
                actions = entity.Actions
                if actions:
                    all_rels.update(actions.column_relationships)
            except Exception as e:  # noqa: BLE001
                await logger.warning(
                    f"Failed to get actions for dataflow {link.id}: {e}"
                )

    # Filter/trace by column name if provided
    if column_name:
        if trace_upstream:
            # Use the converter's trace function
            from ....integrations.graphs.mermaid.column_relationship_erd_converter import (
                _trace_upstream,
            )

            start_entity_id = str(self.parent.id) if self.parent else None
            if start_entity_id:
                all_rels = _trace_upstream(
                    all_rels,
                    start_entity_id=start_entity_id,
                    start_column=column_name,
                )
            else:
                # Fallback: filter to relationships involving the column
                all_rels = {
                    rel
                    for rel in all_rels
                    if rel.from_column == column_name
                    or rel.to_column == column_name
                }
        else:
            # Simple filter: any relationship involving the column
            all_rels = {
                rel
                for rel in all_rels
                if rel.from_column == column_name or rel.to_column == column_name
            }

    return all_rels

to_erd async

to_erd(
    *,
    column_name: str | None = None,
    trace_upstream: bool = True,
    title: str | None = None,
    session: AsyncClient = None,
    debug_api: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
)

Generate a Mermaid ER diagram from column-level lineage.

Convenience method that combines get_column_lineage() with ColumnRelationshipERConverter to produce a visual ERD.

Parameters:

Name	Type	Description	Default
column_name	str | None	Optional column name to trace upstream.	None
trace_upstream	bool	If True and column_name provided, trace only the upstream chain for that column.	True
title	str | None	Optional diagram title.	None
session	AsyncClient	HTTP session for reuse.	None
debug_api	bool	Enable API debug logging.	False
context	RouteContext | None	Route context for API configuration.	None

Returns:

Type	Description
	MermaidERDiagram ready for rendering.

Example

diagram = await card.dataset.Lineage.to_erd( ... column_name="Revenue", ... trace_upstream=True, ... ) print(diagram.to_string())

Source code in src/crew_dcs/classes/subentity/lineage/column_lineage.py

async def to_erd(
    self,
    *,
    column_name: str | None = None,
    trace_upstream: bool = True,
    title: str | None = None,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Generate a Mermaid ER diagram from column-level lineage.

    Convenience method that combines get_column_lineage() with
    ColumnRelationshipERConverter to produce a visual ERD.

    Args:
        column_name: Optional column name to trace upstream.
        trace_upstream: If True and column_name provided, trace only
                       the upstream chain for that column.
        title: Optional diagram title.
        session: HTTP session for reuse.
        debug_api: Enable API debug logging.
        context: Route context for API configuration.

    Returns:
        MermaidERDiagram ready for rendering.

    Example:
        >>> diagram = await card.dataset.Lineage.to_erd(
        ...     column_name="Revenue",
        ...     trace_upstream=True,
        ... )
        >>> print(diagram.to_string())
    """
    from ....integrations.graphs.mermaid import ColumnRelationshipERConverter

    rels = await self.get_column_lineage(
        column_name=column_name,
        trace_upstream=trace_upstream,
        session=session,
        debug_api=debug_api,
        context=context,
        **context_kwargs,
    )

    # Build entity names from lineage
    entity_names: dict[str, str] = {}
    if self.parent:
        entity_names[str(self.parent.id)] = getattr(
            self.parent, "name", getattr(self.parent, "title", str(self.parent.id))
        )
    for link in self.lineage:
        if link.entity:
            entity_names[str(link.id)] = getattr(
                link.entity, "name", getattr(link.entity, "title", str(link.id))
            )

    return ColumnRelationshipERConverter.convert(
        rels,
        entity_names=entity_names,
        title=title
        or f"Column Lineage for {getattr(self.parent, 'name', self.parent.id) if self.parent else 'Unknown'}",
    )