validate_lineage_steps

Public callable

Validate lineage step structure and flag records requiring human review.

Parameters:

Name	Type	Description	Default
lineage_steps	Any	Candidate lineage payload, expected to be a list of dictionaries.	required

Returns:

Type	Description
dict of str to Any	Validation result with is_valid, errors, warnings, and review_required.

Source code in src/fabricops_kit/lineage.py

def validate_lineage_steps(lineage_steps: Any) -> dict[str, Any]:
    """Validate lineage step structure and flag records requiring human review.

    Parameters
    ----------
    lineage_steps : Any
        Candidate lineage payload, expected to be a list of dictionaries.

    Returns
    -------
    dict of str to Any
        Validation result with ``is_valid``, ``errors``, ``warnings``, and ``review_required``.
    """
    errors: list[str] = []
    warnings: list[str] = []
    review_required = False
    required = ["source", "target", "transformation", "reason", "source_type", "target_type", "confidence"]
    if not isinstance(lineage_steps, list):
        return {"is_valid": False, "errors": ["lineage_steps must be a list."], "warnings": [], "review_required": True}
    if not lineage_steps:
        return {"is_valid": False, "errors": ["lineage_steps cannot be empty."], "warnings": [], "review_required": True}
    for i, step in enumerate(lineage_steps, 1):
        if not isinstance(step, dict):
            errors.append(f"Step {i}: each lineage step must be a dict.")
            review_required = True
            continue
        for f in required:
            if f not in step:
                errors.append(f"Step {i}: missing required field '{f}'.")
        if step.get("source_type") == "unknown" or step.get("target_type") == "unknown":
            review_required = True; warnings.append(f"Step {i}: unknown type requires human review.")
        if step.get("confidence") == "low":
            review_required = True; warnings.append(f"Step {i}: low confidence requires human review.")
    return {"is_valid": not errors, "errors": errors, "warnings": warnings, "review_required": review_required}