lineage

lineage

Lineage handlers for Publication and Subscription entities.

DomoLineageLink_Publication dataclass

DomoLineageLink_Publication(
    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/DomoEverywhere/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 DomoPublication

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

    return await DomoPublication.get_by_id(
        publication_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineageLink_Subscription dataclass

DomoLineageLink_Subscription(
    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/DomoEverywhere/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 DomoSubscription

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

    return await DomoSubscription.get_by_id(
        subscription_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineage_Everywhere dataclass

DomoLineage_Everywhere(
    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, ABC

Base class for lineage handlers that manage content collections.

Provides property-based access to categorized lineage links (datasets, cards, pages). Used by both Publication and Subscription lineage handlers.

Important Conceptual Model: - For Publication: datasets/cards/pages are the INPUTS (what was published) - For Subscription: datasets/cards/pages are the INPUTS from publication - Federated entities in subscriber instance are DOWNSTREAM/CHILDREN of subscription (created as a result of subscription, not part of lineage chain)

Properties filter self.lineage (list of DomoLineageLink objects) by type: - datasets: Links with type="DATA_SOURCE" (or legacy "DATASET") - cards: Links with type="CARD" - pages: Links with type="PAGE" - unsorted: Links with unrecognized types

cards property

cards: list[Any]

Get card links from lineage.

Returns:

Type	Description
list[Any]	List of DomoLineageLink objects with type CARD

datasets property

datasets: list[Any]

Get dataset links from lineage.

Returns:

Type	Description
list[Any]	List of DomoLineageLink objects with type DATA_SOURCE (or legacy DATASET)

pages property

pages: list[Any]

Get page links from lineage.

Returns:

Type	Description
list[Any]	List of DomoLineageLink objects with type PAGE

unsorted property

unsorted: list[Any]

Get links with unrecognized types from lineage.

Returns:

Type	Description
list[Any]	List of DomoLineageLink objects that don't match known types

get abstractmethod async

get(
    session: AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth = None,
    parent_auth_retrieval_fn: Callable | None = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Get lineage for the entity - must be implemented by subclasses.

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

@abstractmethod
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 | None = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Get lineage for the entity - must be implemented by subclasses."""
    ...

DomoLineage_Publication dataclass

DomoLineage_Publication(
    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_Everywhere

Lineage handler for publication entities.

The parent is the publication entity this lineage is based off of (not a dependency). Publications contain multiple entities (datasets, cards, pages).

get async

get(
    session: AsyncClient = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth = None,
    parent_auth_retrieval_fn: Callable | None = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Get lineage for all content items in the publication.

Orchestrates fetching lineage for each content item and categorizing the results.

Source code in src/crew_dcs/classes/DomoEverywhere/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 | None = None,
    max_depth: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Get lineage for all content items in the publication.

    Orchestrates fetching lineage for each content item and categorizing the results.
    """
    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,
        )

    # Ensure content is loaded - if not already populated, refetch the publication
    if not self.parent.content:
        from .core import DomoPublication

        refreshed_pub = await DomoPublication.get_by_id(
            publication_id=self.parent.id,
            auth=self.parent.auth,
            context=context,
        )
        self.parent.content = refreshed_pub.content

    if return_raw:
        return self.parent.content

    # Gather lineage for all content items concurrently
    lineage = await dmce.gather_with_concurrency(
        *[
            self._get_content_item_lineage(
                pc=pc,
                context=context,
            )
            for pc in self.parent.content
            if pc
        ],
        n=10,
    )

    # Flatten the list of lists and filter out None values
    result = [ele for sublist in lineage if sublist for ele in sublist]

    # Cache the result for future use
    # Properties (datasets, cards, pages) will filter self.lineage by type
    self.lineage = result
    return result

DomoLineage_Subscription dataclass

DomoLineage_Subscription(
    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_Everywhere

Lineage handler for subscription entities.

Provides categorized access to subscription content via datasets, cards, pages properties.

Conceptual Model: - self.lineage contains: [publication_link] + target_entity_upstream_lineage - datasets/cards/pages properties categorize items from lineage chain - These represent INPUTS to the publication (what was published) - Federated entities in subscriber instance are DOWNSTREAM/CHILDREN of subscription: * They are created as a result of the subscription * They depend ON the subscription (not part of subscription's upstream lineage) * They should be accessed via subscription.get_federated_entities() (if needed)

content property

content

Get the publication content items (what was published/publication inputs).

Returns:

Type	Description
	List of publication content items (PAGE, CARD, DATASET, etc.) that are
	INPUTS to the publication (what was published).

Note

These are NOT the federated entities in the subscriber instance. Federated subscriber entities are DOWNSTREAM/CHILDREN of the subscription and should be accessed separately if needed.

parent_publication property

parent_publication

Get the parent publication entity from the subscription.

Returns:

Type	Description
	DomoPublication entity if available in lineage