lineage

lineage

Lineage handler for Page entities.

DomoLineageLink_Page dataclass

DomoLineageLink_Page(
    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 async

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

Get the entity associated with this lineage link.

Source code in src/crew_dcs/classes/DomoPage/lineage.py

async def get_entity(
    self,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    *,
    context=None,
):
    """Get the entity associated with this lineage link."""
    from .core import DomoPage

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

    return await DomoPage.get_by_id(
        page_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineage_Page dataclass

DomoLineage_Page(
    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(),
    cards: list[Any] = None,
)

Bases: DomoLineage

Lineage handler for page entities.

The parent is the page entity this lineage is based off of (not a dependency). Pages have cards as their immediate dependencies.

get async

get(
    session: AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth = None,
    parent_auth_retrieval_fn: (
        Callable[[str], Any] | None
    ) = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Get lineage for a page by traversing card lineage.

Pages do not have direct datacenter lineage. Instead, we load cards from the page layout and then fetch each card's lineage. Each card is added to the page lineage as a LineageLink with dependencies and dependents referenced from the card's lineage root node.

Source code in src/crew_dcs/classes/DomoPage/lineage.py

async def get(
    self,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth = None,
    parent_auth_retrieval_fn: Callable[[str], Any] | None = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Get lineage for a page by traversing card lineage.

    Pages do not have direct datacenter lineage. Instead, we load cards
    from the page layout and then fetch each card's lineage. Each card
    is added to the page lineage as a LineageLink with dependencies and
    dependents referenced from the card's lineage root node.
    """
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    if not self.parent:
        raise ClassError(
            message="Parent must be set. Use from_parent() to create lineage with a parent.",
            cls_instance=self,
        )

    cache_key = self._build_cache_key(
        parent_auth=parent_auth,
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
        return_raw=return_raw,
    )

    if self.lineage and self._cached_lineage_params == cache_key:
        await logger.debug(
            f"DomoLineage_Page.get: returning cached lineage (len={len(self.lineage)})"
        )
        return list(self.lineage)

    cards = await self.parent.Layout.get_cards(
        context=context,
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
    )

    self.cards = cards

    lineage_cache = self._lineage_cache
    lineage_groups: list[list[DomoLineage_Link]] = []
    card_links: list[DomoLineage_Link] = []

    for card in cards:
        card_lineage = await card.Lineage.get(
            session=session,
            debug_api=debug_api,
            return_raw=return_raw,
            parent_auth=parent_auth,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            max_depth=max_depth,
            lineage_cache=lineage_cache,
            context=context,
        )

        card_root = next(
            (
                link
                for link in card_lineage
                if str(link.id) == str(card.id) and link.type == "CARD"
            ),
            None,
        )

        card_link = DomoLineageLink_Card(
            id=str(card.id),
            auth=card.auth,
            entity=card,
            _type=None,
            dependencies=card_root.dependencies if card_root else [],
            dependents=card_root.dependents if card_root else [],
        )
        card_links.append(card_link)
        lineage_groups.append(card_lineage)

    page_link = self._build_root_link(
        dependencies=card_links,
        dependents=[],
    )

    merged_lineage = self._merge_lineage_links(
        [page_link],
        card_links,
        *lineage_groups,
    )

    self.immediate_dependencies = card_links
    self.immediate_dependents = []
    self.lineage = [page_link] + [
        link for link in merged_lineage if link is not page_link
    ]

    # Connect subscriber page → publisher page if a publisher PAGE
    # exists in the merged lineage (it comes from card federation chains).
    for link in self.lineage:
        if link.type == "PAGE" and str(link.id) != str(page_link.id):
            self._add_direct_link(page_link, link)
            break

    # Connect subscriber page → subscription so the diagram shows the
    # page directly linked to the subscription entity.
    for link in self.lineage:
        if link.type == "SUBSCRIPTION":
            self._add_direct_link(page_link, link)
            break

    self._cached_lineage_params = cache_key

    return list(self.lineage)

get_downstream async

get_downstream(
    session: AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    max_depth: int = 1,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Get downstream lineage for a page.

A page's downstream dependents are publications that include this page. Also includes any datacenter API results.

Source code in src/crew_dcs/classes/DomoPage/lineage.py

async def get_downstream(
    self,
    session: httpx.AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    max_depth: int = 1,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Get downstream lineage for a page.

    A page's downstream dependents are publications that include this page.
    Also includes any datacenter API results.
    """
    from ..DomoEverywhere.lineage import DomoLineageLink_Publication

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    # Try to find publications that include this page
    publication_links: list[DomoLineage_Link] = []
    try:
        # Check if the page is published via the Federation system
        if hasattr(self.parent, "Federation") and self.parent.Federation:
            federation = self.parent.Federation
            if hasattr(federation, "is_published") and federation.is_published:
                # Get the subscription/publication chain
                subscription = getattr(federation, "subscription", None)
                if subscription:
                    publication = getattr(federation, "parent_publication", None)
                    if publication:
                        pub_link = DomoLineageLink_Publication(
                            id=str(publication.id),
                            auth=self.auth,
                            entity=publication,
                            _type="PUBLICATION",
                            dependencies=[],
                            dependents=[],
                        )
                        publication_links.append(pub_link)
    except (AttributeError, ValueError, TypeError) as exc:
        await logger.warning(
            f"DomoLineage_Page.get_downstream: publication discovery failed for page {self.parent.id}: {exc}"
        )

    # Also try the base class datacenter API
    try:
        base_downstream = await super().get_downstream(
            session=session,
            debug_api=debug_api,
            return_raw=return_raw,
            max_depth=max_depth,
            context=context,
            **context_kwargs,
        )
    except (httpx.HTTPStatusError, ValueError) as exc:
        await logger.warning(
            f"DomoLineage_Page.get_downstream: base downstream lookup failed for page {self.parent.id}: {exc}"
        )
        base_downstream = []

    # Merge publication links with base downstream results
    all_links = publication_links + [
        link for link in base_downstream if link not in publication_links
    ]

    self.immediate_dependents = publication_links
    self.downstream_lineage = all_links

    return all_links