publish_lineage

publish_lineage

Publish lineage mixin for DomoLineage.

Handles tracing the publish-lineage chain: subscriber → subscription → publication → publisher entity → publisher lineage

Extracted from base.py for separation of concerns.

DomoLineage_PublishMixin

Publish lineage methods for DomoLineage.

Provides methods for tracing the full publish-lineage chain from subscriber entities through subscriptions and publications to their publisher counterparts.

This mixin is mixed into DomoLineage in base.py.

get_federated_lineage async

get_federated_lineage(
    session: AsyncClient | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    _debug_num_stacks_to_drop: int = 3,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Get lineage for a federated entity.

Always delegates to trace_publish_lineage() which builds the full chain: subscriber → subscription → publication → publisher → publisher lineage.

If the subscription hasn't been discovered yet, ensure_subscription() is called first. If no subscription is found the entity is not published and an empty list is returned.

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

async def get_federated_lineage(
    self,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    return_raw: bool = False,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    _debug_num_stacks_to_drop: int = 3,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Get lineage for a federated entity.

    Always delegates to trace_publish_lineage() which builds the full chain:
    subscriber → subscription → publication → publisher → publisher lineage.

    If the subscription hasn't been discovered yet, ensure_subscription() is
    called first.  If no subscription is found the entity is not published
    and an empty list is returned.
    """
    if not self.parent and (not self.parent_type):
        raise ValueError(
            "Parent must be set. Use from_parent() to create lineage with a parent."
        )

    # Ensure federation context exists
    if not getattr(self.parent, "Federation", None):
        self.parent.enable_federation_support()

    federation = self.parent.Federation

    # Discover subscription if not already known
    if not federation.is_published:
        await federation.ensure_subscription(
            retrieve_parent_auth_fn=parent_auth_retrieval_fn,
            parent_auth=parent_auth,
            entity_type=self.parent.entity_type,
            entity_id=str(self.parent.id),
            session=session,
            debug_api=debug_api,
        )

    if not federation.subscription:
        await logger.warning(
            f"get_federated_lineage: entity {self.parent.id} is federated but no "
            f"subscription found — skipping publisher lineage traversal.",
            entity_id=str(self.parent.id),
            entity_type=self.parent.entity_type,
        )
        return []

    # Delegate to trace_publish_lineage which builds the full chain
    # (subscriber → subscription → publication → publisher → publisher lineage)
    return await self.trace_publish_lineage(
        publish_helper=federation,
        parent_auth=parent_auth,
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
        context=context,
        **context_kwargs,
    )

get_publish_lineage async

get_publish_lineage(
    *,
    publish_helper=None,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Build lineage chain for published entities.

.. deprecated:: Use :meth:trace_publish_lineage instead. This thin wrapper exists only for backward compatibility.

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

async def get_publish_lineage(
    self,
    *,
    publish_helper=None,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Build lineage chain for published entities.

    .. deprecated::
        Use :meth:`trace_publish_lineage` instead.  This thin wrapper
        exists only for backward compatibility.
    """
    return await self.trace_publish_lineage(
        publish_helper=publish_helper,
        parent_auth=parent_auth,
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
        context=context,
        **context_kwargs,
    )

trace_publish_lineage async

trace_publish_lineage(
    *,
    publish_helper=None,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    max_depth: int = 100,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoLineage_Link]

Build the full publish-lineage chain for a federated published entity.

Orchestrates: 1. Ensure subscription is resolved (via FederationContext). 2. Resolve publisher auth. 3. Fetch parent publication. 4. Call _resolve_publisher_entity (overridable hook) to find the publisher-side entity that corresponds to self.parent. 5. Fetch the publisher entity's own lineage. 6. Assemble the chain: subscriber entity → subscription → publication → publisher entity → publisher entity lineage.

Subclasses override _resolve_publisher_entity to customise step 4. For example DomoLineage_Card handles cards that are indirectly published via pages.

Parameters:

Name	Type	Description	Default
publish_helper		FederationContext (defaults to self.Federation).	None
parent_auth	DomoAuth | None	Explicit publisher auth (alternative to retrieval fn).	None
parent_auth_retrieval_fn	Callable | None	Callable(domain) → DomoAuth.	None
max_depth	int	Depth limit for publisher entity lineage.	100
context	RouteContext | None	Route context for API calls.	None

Returns:

Type	Description
list[DomoLineage_Link]	Flat list of lineage links representing the full chain.

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

async def trace_publish_lineage(
    self,
    *,
    publish_helper=None,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    max_depth: int = 100,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoLineage_Link]:
    """Build the full publish-lineage chain for a federated published entity.

    Orchestrates:
    1. Ensure subscription is resolved (via FederationContext).
    2. Resolve publisher auth.
    3. Fetch parent publication.
    4. Call ``_resolve_publisher_entity`` (overridable hook) to find
       the publisher-side entity that corresponds to ``self.parent``.
    5. Fetch the publisher entity's own lineage.
    6. Assemble the chain:
       subscriber entity → subscription → publication → publisher entity
       → publisher entity lineage.

    Subclasses override ``_resolve_publisher_entity`` to customise step 4.
    For example ``DomoLineage_Card`` handles cards that are indirectly
    published via pages.

    Args:
        publish_helper: FederationContext (defaults to ``self.Federation``).
        parent_auth: Explicit publisher auth (alternative to retrieval fn).
        parent_auth_retrieval_fn: Callable(domain) → DomoAuth.
        max_depth: Depth limit for publisher entity lineage.
        context: Route context for API calls.

    Returns:
        Flat list of lineage links representing the full chain.
    """
    from ...DomoEverywhere.lineage import (
        DomoLineageLink_Publication,
        DomoLineageLink_Subscription,
    )

    # --- 1. Ensure subscription ---
    helper = publish_helper or self.Federation
    if not helper:
        raise ValueError(
            "No federation helper available. "
            "Call check_is_published() first or provide publish_helper."
        )

    subscription = await helper.ensure_subscription(
        retrieve_parent_auth_fn=parent_auth_retrieval_fn,
        parent_auth=parent_auth,
        entity_type=self.parent.entity_type,
        entity_id=str(self.parent.id),
        context=context,
        **context_kwargs,
    )
    if not subscription:
        raise DomoError(
            f"Failed to resolve subscription for published entity "
            f"{self.parent.entity_type}:{self.parent.id}"
        )

    # --- 2. Resolve publisher auth ---
    publisher_auth = await helper.get_publisher_auth(
        parent_auth=parent_auth,
        context=context,
        **context_kwargs,
    )

    # --- 3. Fetch parent publication ---
    publication = await helper.get_parent_publication(
        parent_auth=publisher_auth,
        context=context,
        **context_kwargs,
    )

    # --- 4. Resolve publisher entity (overridable hook) ---
    subscriber_domain = subscription.subscriber_domain
    resolution = await self._resolve_publisher_entity(
        publication=publication,
        publisher_auth=publisher_auth,
        subscriber_domain=subscriber_domain,
        context=context,
        **context_kwargs,
    )
    publisher_entity = resolution.entity
    intermediate_entities = resolution.intermediate_entities

    # --- 5. Fetch publisher entity lineage ---
    publisher_entity_lineage: list[DomoLineage_Link] = []
    publisher_entity_link = None

    if publisher_entity:
        publisher_entity_lineage = await publisher_entity.Lineage.get(
            return_raw=False,
            max_depth=max_depth,
            context=context,
            **context_kwargs,
        )

        # Find the publisher entity's own link within its lineage — it
        # already has the correct direct dependencies wired by
        # get_datacenter_lineage().  Only create a new link as a fallback.
        publisher_entity_link = next(
            (
                link
                for link in publisher_entity_lineage
                if str(link.id) == str(publisher_entity.id)
            ),
            None,
        )

        if publisher_entity_link is None:
            link_type = _map_entity_type(publisher_entity.entity_type)
            link_cls = _get_lineage_link_class(link_type)
            publisher_entity_link = link_cls(
                auth=publisher_auth,
                id=str(publisher_entity.id),
                entity=publisher_entity,
                _type=None,
                dependencies=[],
                dependents=[],
            )

    # --- 5b. Build intermediate links (e.g., page for indirect card publication) ---
    # Intermediate entities sit between publication and publisher_entity.
    # Build in reverse so each link's dependencies point to the next in the chain.
    intermediate_links: list[DomoLineage_Link] = []
    next_dependency = publisher_entity_link

    for intermediate_entity in reversed(intermediate_entities):
        inter_link_type = _map_entity_type(intermediate_entity.entity_type)
        inter_link_cls = _get_lineage_link_class(inter_link_type)
        inter_link = inter_link_cls(
            auth=publisher_auth,
            id=str(intermediate_entity.id),
            entity=intermediate_entity,
            _type=None,
            dependencies=[next_dependency] if next_dependency else [],
            dependents=[],
        )
        if next_dependency:
            next_dependency.dependents = [inter_link]
        intermediate_links.insert(0, inter_link)
        next_dependency = inter_link

    # Determine what the publication link should point to
    publication_first_dep = (
        intermediate_links[0] if intermediate_links else publisher_entity_link
    )

    # --- 6. Assemble the chain ---
    publication_link = DomoLineageLink_Publication(
        auth=publisher_auth,
        id=str(publication.id),
        entity=publication,
        _type="PUBLICATION",
        dependents=[],
        dependencies=([publication_first_dep] if publication_first_dep else []),
    )

    subscription_link = DomoLineageLink_Subscription(
        auth=self.parent.auth,
        id=str(subscription.id),
        entity=subscription,
        _type="SUBSCRIPTION",
        dependents=[],
        dependencies=[publication_link],
    )

    subscriber_link = _get_lineage_link_class(
        _map_entity_type(self.parent.entity_type)
    )(
        auth=self.parent.auth,
        id=str(self.parent.id),
        entity=self.parent,
        _type=None,
        dependencies=[subscription_link],
        dependents=[],
    )

    # Wire back-pointers
    subscription_link.dependents = [subscriber_link]
    publication_link.dependents = [subscription_link]
    if publication_first_dep:
        publication_first_dep.dependents = [publication_link]

    chain = [
        subscriber_link,
        subscription_link,
        publication_link,
    ]
    chain.extend(intermediate_links)
    # publisher_entity_link is already within publisher_entity_lineage
    # (found by id match), so only extend — don't append separately.
    chain.extend(publisher_entity_lineage)

    return chain