ai_readiness_lineage

ai_readiness_lineage

AI Readiness lineage auto-fill for DomoDataset.

Provides lineage-aware auto-population of AI Readiness assessments by tracing upstream and downstream dependencies and generating human-readable context descriptions.

Data Models

LineageEntityInfo: Summary of a lineage entity for AI Readiness AIReadinessLineageReport: Complete lineage report for an AI Readiness assessment

Core Functions

generate_lineage_report: Trace upstream and downstream lineage for a dataset auto_fill_lineage_context: Auto-fill AI Readiness context based on lineage batch_populate_lineage: Populate lineage for multiple datasets in parallel

AIReadinessLineageReport dataclass

AIReadinessLineageReport(
    dataset_id: str,
    dataset_name: str,
    upstream_datasets: list[LineageEntityInfo] = list(),
    upstream_dataflows: list[LineageEntityInfo] = list(),
    downstream_cards: list[LineageEntityInfo] = list(),
    downstream_datasets: list[LineageEntityInfo] = list(),
    downstream_pages: list[LineageEntityInfo] = list(),
)

Complete lineage report for an AI Readiness assessment.

Attributes:

Name	Type	Description
dataset_id	str	ID of the dataset this report is for
dataset_name	str	Display name of the dataset
upstream_datasets	list[LineageEntityInfo]	Source datasets that feed into this dataset
upstream_dataflows	list[LineageEntityInfo]	Dataflows that produce this dataset
downstream_cards	list[LineageEntityInfo]	Cards that depend on this dataset
downstream_datasets	list[LineageEntityInfo]	Downstream datasets (e.g., views) that depend on this dataset
downstream_pages	list[LineageEntityInfo]	Pages that contain cards using this dataset

to_description

to_description() -> str

Generate a natural language description for AI Readiness context field.

Produces a human-readable paragraph summarizing the dataset's lineage, suitable for use as the AI Readiness description field.

Returns:

Type	Description
str	A paragraph describing the dataset's lineage context

Source code in src/crew_dcs/classes/DomoDataset/ai_readiness_lineage.py

def to_description(self) -> str:  # noqa: C901
    """Generate a natural language description for AI Readiness context field.

    Produces a human-readable paragraph summarizing the dataset's lineage,
    suitable for use as the AI Readiness description field.

    Returns:
        A paragraph describing the dataset's lineage context
    """
    parts: list[str] = []

    # --- Upstream context ---
    if self.upstream_datasets or self.upstream_dataflows:
        upstream_parts: list[str] = []

        if self.upstream_datasets:
            ds_names = [ds.name for ds in self.upstream_datasets[:5]]
            ds_label = ", ".join(ds_names)
            if len(self.upstream_datasets) > 5:
                ds_label += f" and {len(self.upstream_datasets) - 5} more"
            upstream_parts.append(
                f"dataset{'s' if len(self.upstream_datasets) > 1 else ''} {ds_label}"
            )

        if self.upstream_dataflows:
            df_names = [df.name for df in self.upstream_dataflows[:5]]
            df_label = ", ".join(df_names)
            if len(self.upstream_dataflows) > 5:
                df_label += f" and {len(self.upstream_dataflows) - 5} more"
            upstream_parts.append(
                f"dataflow{'s' if len(self.upstream_dataflows) > 1 else ''} {df_label}"
            )

        if len(upstream_parts) == 1:
            parts.append(f"This dataset is sourced from {upstream_parts[0]}")
        elif len(upstream_parts) == 2:
            parts.append(
                f"This dataset is sourced from {upstream_parts[0]} via {upstream_parts[1]}"
            )
    else:
        parts.append("This dataset has no upstream dependencies")

    # --- Downstream context ---
    downstream_parts: list[str] = []

    if self.downstream_cards:
        card_count = len(self.downstream_cards)
        downstream_parts.append(f"{card_count} card{'s' if card_count > 1 else ''}")

    if self.downstream_datasets:
        ds_count = len(self.downstream_datasets)
        ds_names = [ds.name for ds in self.downstream_datasets[:3]]
        ds_label = ", ".join(ds_names)
        if ds_count > 3:
            ds_label += f" and {ds_count - 3} more"
        downstream_parts.append(
            f"{ds_count} downstream dataset{'s' if ds_count > 1 else ''} including {ds_label}"
        )

    if self.downstream_pages:
        page_count = len(self.downstream_pages)
        downstream_parts.append(f"{page_count} page{'s' if page_count > 1 else ''}")

    if downstream_parts:
        parts.append(f"and feeds into {' and '.join(downstream_parts)}")
    else:
        if parts[-1] != "This dataset has no upstream dependencies":
            parts.append("and has no downstream dependents")
        else:
            parts.append("and has no downstream dependents")

    return ". ".join(parts) + "."

to_dict

to_dict() -> dict

Convert to dictionary format.

Source code in src/crew_dcs/classes/DomoDataset/ai_readiness_lineage.py

def to_dict(self) -> dict:
    """Convert to dictionary format."""
    return {
        "datasetId": self.dataset_id,
        "datasetName": self.dataset_name,
        "upstreamDatasets": [e.to_dict() for e in self.upstream_datasets],
        "upstreamDataflows": [e.to_dict() for e in self.upstream_dataflows],
        "downstreamCards": [e.to_dict() for e in self.downstream_cards],
        "downstreamDatasets": [e.to_dict() for e in self.downstream_datasets],
        "downstreamPages": [e.to_dict() for e in self.downstream_pages],
    }

LineageEntityInfo dataclass

LineageEntityInfo(
    id: str,
    name: str,
    type: str,
    description: str | None = None,
)

Summary of a lineage entity for AI Readiness.

Attributes:

Name	Type	Description
id	str	Entity identifier
name	str	Display name of the entity
type	str	Entity type string (DATA_SOURCE, DATAFLOW, CARD, PAGE)
description	str | None	Optional description of the entity

to_dict

to_dict() -> dict

Convert to dictionary format.

Source code in src/crew_dcs/classes/DomoDataset/ai_readiness_lineage.py

def to_dict(self) -> dict:
    """Convert to dictionary format."""
    return {
        "id": self.id,
        "name": self.name,
        "type": self.type,
        "description": self.description,
    }

auto_fill_lineage_context async

auto_fill_lineage_context(
    auth: DomoAuth,
    dataset_id: str,
    max_depth: int = 5,
    *,
    context: RouteContext | None = None
) -> DomoDataset_AI_Readiness

Auto-fill AI Readiness context based on lineage.

1. Generate lineage report
2. Get or create AI Readiness assessment
3. Update the description field with lineage context
4. Update column descriptions where possible (e.g., mark columns as "upstream from {dataset_name}")
5. Return the updated AI Readiness object

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
dataset_id	str	ID of the dataset	required
max_depth	int	Maximum lineage depth for upstream traversal (default: 5)	5
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
DomoDataset_AI_Readiness	Updated DomoDataset_AI_Readiness with lineage context populated

Source code in src/crew_dcs/classes/DomoDataset/ai_readiness_lineage.py

async def auto_fill_lineage_context(
    auth: DomoAuth,
    dataset_id: str,
    max_depth: int = 5,
    *,
    context: RouteContext | None = None,
) -> DomoDataset_AI_Readiness:
    """Auto-fill AI Readiness context based on lineage.

    1. Generate lineage report
    2. Get or create AI Readiness assessment
    3. Update the description field with lineage context
    4. Update column descriptions where possible (e.g., mark columns as
       "upstream from {dataset_name}")
    5. Return the updated AI Readiness object

    Args:
        auth: Authentication object
        dataset_id: ID of the dataset
        max_depth: Maximum lineage depth for upstream traversal (default: 5)
        context: Optional RouteContext for API call configuration

    Returns:
        Updated DomoDataset_AI_Readiness with lineage context populated
    """
    from .core import DomoDataset

    # Step 1: Generate lineage report
    report = await generate_lineage_report(
        auth=auth,
        dataset_id=dataset_id,
        max_depth=max_depth,
        include_downstream=True,
        context=context,
    )

    # Step 2: Fetch the dataset and get/create AI Readiness
    dataset = await DomoDataset.get_by_id(
        auth=auth,
        dataset_id=dataset_id,
        is_get_account=False,
        context=context,
    )

    ai_readiness = dataset.AI_Readiness
    if ai_readiness is None:
        # Shouldn't happen since __post_init__ creates it, but be safe
        ai_readiness = DomoDataset_AI_Readiness.from_parent(parent=dataset)

    # Try to get existing AI readiness; if not found, create one
    try:
        await ai_readiness.get(context=context)
    except (DomoError, httpx.HTTPStatusError):
        await ai_readiness.create(
            dictionary_name=dataset.name or dataset_id,
            description=report.to_description(),
            context=context,
        )
        return ai_readiness

    # Step 3: Update the description field with lineage context
    lineage_description = report.to_description()

    # If there's an existing description, append lineage context
    if ai_readiness.description:
        updated_description = (
            f"{ai_readiness.description}\n\nLineage Context: {lineage_description}"
        )
    else:
        updated_description = lineage_description

    # Step 4: Update column descriptions where possible
    # Mark columns from upstream datasets
    columns = ai_readiness.columns
    for col in columns:
        if not col.description and report.upstream_datasets:
            # Use the first upstream dataset name as context
            primary_source = report.upstream_datasets[0].name
            col.description = f"Source column from {primary_source}"

    # Step 5: Update via API
    await ai_readiness.update(
        description=updated_description,
        columns=columns,
        context=context,
    )

    return ai_readiness

batch_populate_lineage async

batch_populate_lineage(
    auth: DomoAuth,
    dataset_ids: list[str],
    max_depth: int = 5,
    *,
    context: RouteContext | None = None
) -> dict[str, AIReadinessLineageReport]

Populate lineage for multiple datasets in parallel.

Uses gather_with_concurrency for efficient batch processing. Returns dict mapping dataset_id to lineage report.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
dataset_ids	list[str]	List of dataset IDs to process	required
max_depth	int	Maximum lineage depth for upstream traversal (default: 5)	5
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
dict[str, AIReadinessLineageReport]	Dictionary mapping dataset_id to AIReadinessLineageReport

Source code in src/crew_dcs/classes/DomoDataset/ai_readiness_lineage.py

async def batch_populate_lineage(
    auth: DomoAuth,
    dataset_ids: list[str],
    max_depth: int = 5,
    *,
    context: RouteContext | None = None,
) -> dict[str, AIReadinessLineageReport]:
    """Populate lineage for multiple datasets in parallel.

    Uses gather_with_concurrency for efficient batch processing.
    Returns dict mapping dataset_id to lineage report.

    Args:
        auth: Authentication object
        dataset_ids: List of dataset IDs to process
        max_depth: Maximum lineage depth for upstream traversal (default: 5)
        context: Optional RouteContext for API call configuration

    Returns:
        Dictionary mapping dataset_id to AIReadinessLineageReport
    """
    results = await dmce.gather_with_concurrency(
        *[
            generate_lineage_report(
                auth=auth,
                dataset_id=dataset_id,
                max_depth=max_depth,
                include_downstream=True,
                context=context,
            )
            for dataset_id in dataset_ids
        ],
        n=5,
    )

    return {report.dataset_id: report for report in results}

generate_lineage_report async

generate_lineage_report(
    auth: DomoAuth,
    dataset_id: str,
    max_depth: int = 5,
    include_downstream: bool = True,
    *,
    context: RouteContext | None = None
) -> AIReadinessLineageReport

Trace upstream and downstream lineage for a dataset.

1. Fetch the dataset metadata for name
2. Get upstream lineage (source datasets and dataflows)
3. Get downstream lineage (dependent cards, datasets, pages)
4. Return structured report

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object	required
dataset_id	str	ID of the dataset to trace	required
max_depth	int	Maximum lineage depth for upstream traversal (default: 5)	5
include_downstream	bool	Whether to include downstream dependents (default: True)	True
context	RouteContext | None	Optional RouteContext for API call configuration	None

Returns:

Type	Description
AIReadinessLineageReport	AIReadinessLineageReport with categorized lineage information

Source code in src/crew_dcs/classes/DomoDataset/ai_readiness_lineage.py

async def generate_lineage_report(
    auth: DomoAuth,
    dataset_id: str,
    max_depth: int = 5,
    include_downstream: bool = True,
    *,
    context: RouteContext | None = None,
) -> AIReadinessLineageReport:
    """Trace upstream and downstream lineage for a dataset.

    1. Fetch the dataset metadata for name
    2. Get upstream lineage (source datasets and dataflows)
    3. Get downstream lineage (dependent cards, datasets, pages)
    4. Return structured report

    Args:
        auth: Authentication object
        dataset_id: ID of the dataset to trace
        max_depth: Maximum lineage depth for upstream traversal (default: 5)
        include_downstream: Whether to include downstream dependents (default: True)
        context: Optional RouteContext for API call configuration

    Returns:
        AIReadinessLineageReport with categorized lineage information
    """
    from .core import DomoDataset

    # Fetch dataset to get the name
    try:
        dataset = await DomoDataset.get_by_id(
            auth=auth,
            dataset_id=dataset_id,
            is_get_account=False,
            context=context,
        )
        dataset_name = dataset.name or dataset_id
    except (DomoError, httpx.HTTPStatusError) as e:
        await logger.warning(
            f"Failed to fetch dataset {dataset_id} for lineage report: {e}"
        )
        dataset_name = dataset_id

    # Fetch upstream lineage
    upstream_links = await _fetch_upstream_lineage(
        auth=auth,
        dataset_id=dataset_id,
        max_depth=max_depth,
        context=context,
    )

    # Fetch downstream lineage (if requested)
    downstream_links: list[Any] = []
    if include_downstream:
        downstream_links = await _fetch_downstream_lineage(
            auth=auth,
            dataset_id=dataset_id,
            context=context,
        )

    # Categorize links
    categorized = _categorize_links(upstream_links, downstream_links)

    return AIReadinessLineageReport(
        dataset_id=dataset_id,
        dataset_name=dataset_name,
        upstream_datasets=categorized["upstream_datasets"],
        upstream_dataflows=categorized["upstream_dataflows"],
        downstream_cards=categorized["downstream_cards"],
        downstream_datasets=categorized["downstream_datasets"],
        downstream_pages=categorized["downstream_pages"],
    )