Skip to content

lineage

lineage

Lineage handler for Dataset entities.

DomoLineageLink_Dataset(
    auth: DomoAuth,
    id: str,
    entity: Any,
    _type: str | None = None,
    dependencies: list[DomoLineage_Link] = list(),
    dependents: list[DomoLineage_Link] = list(),
)

Bases: DomoLineage_Link

get_entity(
    debug_api: bool = False,
    session: AsyncClient | None = None,
    *,
    context=None
)

Get the entity associated with this lineage link. This method should be implemented by subclasses to return the appropriate entity. Uses DomoDataset factory to automatically detect views, federated datasets, etc.

Source code in src/crew_dcs/classes/DomoDataset/lineage.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
async def get_entity(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context=None,
):
    """
    Get the entity associated with this lineage link.
    This method should be implemented by subclasses to return the appropriate entity.
    Uses DomoDataset factory to automatically detect views, federated datasets, etc.
    """
    from ...client.context import RouteContext
    from .core import DomoDataset

    if context is None:
        context = RouteContext(
            session=session,
            debug_api=debug_api,
        )

    return await DomoDataset.get_by_id(
        dataset_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineage_Dataset dataclass 

DomoLineage_Dataset(
    auth: DomoAuth,
    parent: Any = None,
    lineage: list[DomoLineage_Link] = list(),
    immediate_dependencies: list[DomoLineage_Link] = list(),
    immediate_dependents: list[DomoLineage_Link] = list(),
    downstream_lineage: list[DomoLineage_Link] = list(),
)

Bases: DomoLineage

Lineage handler for dataset entities.

The parent is the dataset entity this lineage is based off of (not a dependency). For dataset views, this fetches the parent dataset. For raw datasets, this fetches upstream dependencies (if any).

get_parent_dataset 

get_parent_dataset() -> Any | None

Get the parent dataset entity from lineage for dataset views.

For dataset views, the parent dataset is included in the lineage as a DATA_SOURCE. This method extracts it from the lineage. For non-view datasets, returns None.

Returns:

Type Description
Any | None

The parent dataset entity instance, or None if not found or not a view
Example

lineage = await dataset_view.Lineage.get(max_depth=1) parent = dataset_view.Lineage.get_parent_dataset() if parent: ... print(f"Parent dataset: {parent.name}")

Source code in src/crew_dcs/classes/DomoDataset/lineage.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def get_parent_dataset(self) -> Any | None:
    """Get the parent dataset entity from lineage for dataset views.

    For dataset views, the parent dataset is included in the lineage as a DATA_SOURCE.
    This method extracts it from the lineage. For non-view datasets, returns None.

    Returns:
        The parent dataset entity instance, or None if not found or not a view

    Example:
        >>> lineage = await dataset_view.Lineage.get(max_depth=1)
        >>> parent = dataset_view.Lineage.get_parent_dataset()
        >>> if parent:
        ...     print(f"Parent dataset: {parent.name}")
    """
    # Only applicable for dataset views
    if not (hasattr(self.parent, "is_view") and self.parent.is_view):
        return None

    # The parent dataset should be the DATA_SOURCE in the lineage
    # (views typically have one parent dataset)
    for item in self.lineage:
        if item.type == "DATA_SOURCE" and hasattr(item, "entity") and item.entity:
            return item.entity

    return None