check_schema_drift

Public callable

Compare a current dataframe schema against a baseline schema snapshot.

Parameters:

Name	Type	Description	Default
df	Any	Value used by this callable.	required
dataset_name	Any	Value used by this callable.	required
table_name	Any	Value used by this callable.	required
baseline_snapshot	Any	Value used by this callable.	None
policy	Any	Value used by this callable.	None
engine	Any	Value used by this callable.	'spark'

Returns:

Type	Description
dict	Structured output produced by this callable.

Source code in src/fabricops_kit/drift.py

def check_schema_drift(df, dataset_name: str, table_name: str, baseline_snapshot: dict | None = None, policy: dict | None = None, engine: str = "spark") -> dict:
    """Compare a current dataframe schema against a baseline schema snapshot.

        Parameters
        ----------
        df : Any
            Value used by this callable.
        dataset_name : Any
            Value used by this callable.
        table_name : Any
            Value used by this callable.
        baseline_snapshot : Any
            Value used by this callable.
        policy : Any
            Value used by this callable.
        engine : Any
            Value used by this callable.

        Returns
        -------
        dict
            Structured output produced by this callable.
    """
    current_snapshot = build_schema_snapshot(df, dataset_name=dataset_name, table_name=table_name, engine=engine)
    if baseline_snapshot is None:
        return {
            "dataset_name": dataset_name,
            "table_name": table_name,
            "status": "no_baseline",
            "can_continue": True,
            "current_snapshot": current_snapshot,
            "baseline_snapshot": None,
            "comparison": None,
            "message": "No baseline schema snapshot found; current snapshot captured as first observation.",
        }

    comparison = compare_schema_snapshots(baseline_snapshot, current_snapshot, policy=policy or default_schema_drift_policy())
    status = str(comparison.get("status", "passed"))
    return {
        "dataset_name": dataset_name,
        "table_name": table_name,
        "status": status,
        "can_continue": bool(comparison.get("can_continue", True)),
        "current_snapshot": current_snapshot,
        "baseline_snapshot": baseline_snapshot,
        "comparison": comparison,
        "message": "Schema drift check completed.",
    }