lineage_link

lineage_link

Lineage link base and registry for dependency graph construction.

Entity-specific lineage link subclasses are defined alongside their entities (for example classes/DomoDataset/lineage.py).

DomoLineage_Link dataclass

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

Bases: DomoBase

Represents a link in the dependency lineage graph.

Terminology: Uses dependency graph terminology:

• dependencies: Upstream dependencies (what this entity depends on) Example: A card's dependencies are the datasets it uses for data Example: A view's dependencies are the datasets it depends on

• dependents: Downstream dependents (what depends on this entity) Example: A dataset's dependents are the cards/pages/views that use it

This matches standard dependency graph terminology and the Domo API structure. Note: This is different from composition (e.g., a page "contains" cards).

The datacenter lineage payload frequently references entities only by ID/type; we therefore construct placeholder links with entity=None and materialize the real entity later via :meth:get_entity. The _type cache allows those placeholder links to behave consistently (hashing, comparisons, rendering) until their backing entity is fetched.

type property

type: str

Get the lineage type, derived from entity if available, otherwise from cached _type.

from_dict classmethod

from_dict(
    lineage_api_obj: dict[str, Any], auth: DomoAuth
) -> DomoLineage_Link

Create a DomoLineage_Link from a lineage API object dict. Gets the appropriate link class based on the 'type' field.

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

@classmethod
def from_dict(
    cls,
    lineage_api_obj: dict[str, Any],
    auth: DomoAuth,
) -> DomoLineage_Link:
    """
    Create a DomoLineage_Link from a lineage API object dict.
    Gets the appropriate link class based on the 'type' field.
    """
    # Get the specific link class for this type
    link_cls = _get_lineage_link_class(lineage_api_obj["type"])

    return link_cls(
        auth=auth,
        id=lineage_api_obj["id"],
        entity=None,  # Placeholder - entity not loaded yet
        _type=lineage_api_obj["type"],
        dependents=cls._create_lineage_links_from_dicts(
            lineage_api_obj.get("children", []), auth=auth
        ),
        dependencies=cls._create_lineage_links_from_dicts(
            lineage_api_obj.get("parents", []), auth=auth
        ),
    )

get_entity abstractmethod async

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

Get the entity associated with this lineage link. Uses self.id and self.auth to fetch the entity.

This method should be implemented by subclasses to return the appropriate entity.

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

@abstractmethod
async def get_entity(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context=None,
):
    """
    Get the entity associated with this lineage link.
    Uses self.id and self.auth to fetch the entity.

    This method should be implemented by subclasses to return the appropriate entity.
    """
    raise NotImplementedError("Subclasses must implement this method.")

register_lineage_link

register_lineage_link(link_type: str)

Decorator to register a DomoLineage_Link subclass by datacenter type string.

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

def register_lineage_link(link_type: str):
    """Decorator to register a DomoLineage_Link subclass by datacenter type string."""

    def decorator(cls: type[DomoLineage_Link]) -> type[DomoLineage_Link]:
        _LINEAGE_LINK_REGISTRY[link_type] = cls
        return cls

    return decorator

resolve_relationship

resolve_relationship(from_type: str, to_type: str) -> str

Determine the semantic relationship type between two lineage nodes.

Uses normalised entity types (DATA_SOURCE, DATAFLOW, CARD, etc.) to pick the most descriptive label for the Neo4j edge.

Rules (evaluated in order): SUBSCRIPTION → PUBLICATION : MANAGED_BY DATA_SOURCE/CARD/PAGE → SUBSCRIPTION : MANAGED_BY DATA_SOURCE → DATAFLOW : CREATED_BY CARD → DATA_SOURCE : QUERIED_BY PAGE → CARD : CONTAINS everything else : DEPENDS_ON

Parameters:

Name	Type	Description	Default
from_type	str	Normalised entity type of the dependent (downstream) node.	required
to_type	str	Normalised entity type of the dependency (upstream) node.	required

Returns:

Type	Description
str	A relationship label string from ALLOWED_RELATIONSHIPS.

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

def resolve_relationship(from_type: str, to_type: str) -> str:
    """Determine the semantic relationship type between two lineage nodes.

    Uses normalised entity types (DATA_SOURCE, DATAFLOW, CARD, etc.) to pick the
    most descriptive label for the Neo4j edge.

    Rules (evaluated in order):
        SUBSCRIPTION  → PUBLICATION  : MANAGED_BY
        DATA_SOURCE/CARD/PAGE → SUBSCRIPTION : MANAGED_BY
        DATA_SOURCE   → DATAFLOW     : CREATED_BY
        CARD          → DATA_SOURCE  : QUERIED_BY
        PAGE          → CARD         : CONTAINS
        everything else              : DEPENDS_ON

    Args:
        from_type: Normalised entity type of the dependent (downstream) node.
        to_type:   Normalised entity type of the dependency (upstream) node.

    Returns:
        A relationship label string from ``ALLOWED_RELATIONSHIPS``.
    """
    # Subscription managed by publication
    if from_type == "SUBSCRIPTION" and to_type == "PUBLICATION":
        return "MANAGED_BY"

    # Subscriber entities managed by their subscription
    if from_type in {"DATA_SOURCE", "CARD", "PAGE"} and to_type == "SUBSCRIPTION":
        return "MANAGED_BY"

    # Dataflow creates datasets
    if from_type == "DATA_SOURCE" and to_type == "DATAFLOW":
        return "CREATED_BY"

    # Cards query datasets
    if from_type == "CARD" and to_type == "DATA_SOURCE":
        return "QUERIED_BY"

    # Page contains cards
    if from_type == "PAGE" and to_type == "CARD":
        return "CONTAINS"

    return "DEPENDS_ON"