build_lineage_records

Public callable

Build compact lineage records for downstream metadata sinks.

Parameters:

Name	Type	Description	Default
dataset_name	str	Dataset identifier for all output rows.	required
run_id	str	Unique run identifier.	required
source_tables	list of str	Source table names captured for the run.	required
target_table	str	Target table name produced by the run.	required
transformation_steps	list of dict	Transformation step dictionaries to merge into each output row.	required

Returns:

Type	Description
list of dict	Row dictionaries suitable for metadata persistence.

Source code in src/fabricops_kit/lineage.py

def build_lineage_records(*, dataset_name: str, run_id: str, source_tables: list[str], target_table: str, transformation_steps: list[dict]) -> list[dict]:
    """Build compact lineage records for downstream metadata sinks.

    Parameters
    ----------
    dataset_name : str
        Dataset identifier for all output rows.
    run_id : str
        Unique run identifier.
    source_tables : list of str
        Source table names captured for the run.
    target_table : str
        Target table name produced by the run.
    transformation_steps : list of dict
        Transformation step dictionaries to merge into each output row.

    Returns
    -------
    list of dict
        Row dictionaries suitable for metadata persistence.
    """
    return [{"run_id": run_id, "dataset_name": dataset_name, "source_tables": source_tables, "target_table": target_table, **s} for s in transformation_steps]