IVYXSTUDIO · IVY NODE

Rows Validate

ivy.node.rows-validate · v0.1.0

ivyx✓

Checks every row against a JSON Schema and splits the rows into valid ones and invalid ones with their problems. Returns has_errors so a flow can branch on it.

#validate#schema#data#quality

Inputs

Field	Type	Description
rowsrequired	array	The rows to check.
schemarequired	object	A JSON Schema for one row: type, required, properties, enum, minimum, maximum, minLength.

Outputs

Field	Type	Description
valid_rowsrequired	array	Rows with no problem.
invalid_rowsrequired	array	Rows with problems: index, the row's own columns, and problems as one sentence.
error_countrequired	integer	How many rows had a problem.
has_errorsrequired	boolean	True when any row had a problem.

Source

python

inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]

rows = inp["rows"]
schema = inp["schema"]

PY = {"object": dict, "array": list, "string": str, "number": (int, float), "integer": int, "boolean": bool}


def type_of(value):
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, int):
        return "integer"
    if isinstance(value, float):
        return "number"
    return {dict: "object", list: "array", str: "string"}.get(type(value), type(value).__name__)


def check(value, schema, path):
    problems = []
    want = schema.get("type")
    if want:
        allowed = want if isinstance(want, list) else [want]
        got = type_of(value)
        if not (got in allowed or (got == "integer" and "number" in allowed)):
            return [f"{path}: expected {'/'.join(allowed)}, got {got}"]
    if "enum" in schema and value not in schema["enum"]:
        problems.append(f"{path}: {value!r} is not one of {schema['enum']}")
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if "minimum" in schema and value < schema["minimum"]:
            problems.append(f"{path}: {value} is below {schema['minimum']}")
        if "maximum" in schema and value > schema["maximum"]:
            problems.append(f"{path}: {value} is above {schema['maximum']}")
    if isinstance(value, str) and "minLength" in schema and len(value) < schema["minLength"]:
        problems.append(f"{path}: shorter than {schema['minLength']}")
    if isinstance(value, dict):
        for key in schema.get("required", []):
            if key not in value or value[key] is None:
                problems.append(f"{path}.{key}: required")
        for key, sub in schema.get("properties", {}).items():
            if key in value and value[key] is not None:
                problems.extend(check(value[key], sub, f"{path}.{key}"))
    return problems


valid, invalid = [], []
for index, row in enumerate(rows):
    problems = check(row, schema, "$")
    if problems:
        # Flat, so a report table shows the row's own columns beside why.
        invalid.append({"index": index, **row, "problems": "; ".join(problems)})
    else:
        valid.append(row)

out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out["valid_rows"] = valid
out["invalid_rows"] = invalid
out["error_count"] = len(invalid)
out["has_errors"] = bool(invalid)

Tests

Requires: python:3.9

splits
A row missing its amount and a negative one are both reported.
all-valid
Clean rows report no errors.