Skip to content

build_schema_snapshot_records

Public callable

Convert a schema snapshot into row-wise metadata records.

Parameters:

Name Type Description Default
snapshot Any

Value used by this callable.

 required

Returns:

Type Description
list[dict]

Structured output produced by this callable.
Source code in src/fabricops_kit/metadata.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def build_schema_snapshot_records(snapshot: dict, *, run_id: str, table_stage: str) -> list[dict]:
    """Convert a schema snapshot into row-wise metadata records.

        Parameters
        ----------
        snapshot : Any
            Value used by this callable.

        Returns
        -------
        list[dict]
            Structured output produced by this callable.
    """
    base = {
        "run_id": run_id,
        "dataset_name": snapshot.get("dataset_name"),
        "table_name": snapshot.get("table_name"),
        "table_stage": table_stage,
        "engine": snapshot.get("engine"),
        "generated_at": snapshot.get("generated_at"),
    }
    return [
        to_jsonable(
            {
                **base,
                "column_name": col.get("column_name"),
                "ordinal_position": col.get("ordinal_position"),
                "data_type": col.get("data_type"),
                "nullable": col.get("nullable"),
                "column_hash": col.get("column_hash"),
            }
        )
        for col in snapshot.get("columns", [])
    ]