lineage

lineage

Lineage handler for Card entities.

DomoLineageLink_Card dataclass

DomoLineageLink_Card(
    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/DomoCard/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 DomoCard

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

    return await DomoCard.get_by_id(
        card_id=self.id,
        auth=self.auth,
        context=context,
    )

DomoLineage_Card dataclass

DomoLineage_Card(
    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 card entities.

Overrides _resolve_publisher_entity to handle indirect publication: cards that are published as part of a page rather than directly.

validate_datasources

validate_datasources() -> dict[str, Any]

Validate that this card has at most one datasource.

CustomApp/EnterpriseApp cards (identified by having a datastore_id) are allowed to have multiple datasources. Regular cards should have at most one.

Returns:

Type	Description
dict[str, Any]	Dictionary with validation results:
dict[str, Any]	valid: bool - True if card has valid datasource count
dict[str, Any]	datasource_count: int - Number of datasources
dict[str, Any]	is_custom_app: bool - True if card is part of CustomApp/EnterpriseApp
dict[str, Any]	violation: dict | None - Violation details if invalid, None otherwise

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

def validate_datasources(self) -> dict[str, Any]:
    """Validate that this card has at most one datasource.

    CustomApp/EnterpriseApp cards (identified by having a datastore_id) are allowed
    to have multiple datasources. Regular cards should have at most one.

    Returns:
        Dictionary with validation results:
        - valid: bool - True if card has valid datasource count
        - datasource_count: int - Number of datasources
        - is_custom_app: bool - True if card is part of CustomApp/EnterpriseApp
        - violation: dict | None - Violation details if invalid, None otherwise
    """
    # Count DATA_SOURCE items in lineage
    datasource_count = sum(1 for item in self.lineage if item.type == "DATA_SOURCE")

    # Check if card is part of CustomApp/EnterpriseApp
    # Cards with datastore_id are CustomApp/EnterpriseApp cards
    is_custom_app = bool(getattr(self.parent, "datastore_id", None))

    # CustomApp/EnterpriseApp cards can have multiple datasources
    # Regular cards should have at most one
    max_allowed = float("inf") if is_custom_app else 1
    is_valid = datasource_count <= max_allowed

    result = {
        "valid": is_valid,
        "datasource_count": datasource_count,
        "is_custom_app": is_custom_app,
        "violation": None,
    }

    if not is_valid:
        # Use entity_name property if available
        if hasattr(self.parent, "entity_name"):
            card_name = self.parent.entity_name
        else:
            card_name = (
                getattr(self.parent, "title", None)
                or getattr(self.parent, "name", None)
                or str(self.parent.id)
            )

        result["violation"] = {
            "card_id": self.parent.id,
            "card_name": card_name,
            "datasource_count": datasource_count,
            "max_allowed": max_allowed,
            "datasources": [
                {
                    "id": item.id,
                    "name": (
                        item.entity.entity_name
                        if item.entity and hasattr(item.entity, "entity_name")
                        else (
                            getattr(item.entity, "name", item.id)
                            if item.entity
                            else item.id
                        )
                    ),
                }
                for item in self.lineage
                if item.type == "DATA_SOURCE"
            ],
        }

    return result