Data Validator

Data Validator

Why this skill exists

Data validation is tedious and error-prone to do manually. This skill automates common validation patterns — type checking, format validation, referential integrity, business rules — and produces a clear validation report.

When to use

• User has a dataset and wants to verify it's correct before processing or importing
• User wants to find bad rows (nulls, invalid emails, wrong formats, out-of-range values)
• User wants to validate JSON against a schema or CSV against expected column rules

Step-by-step procedure

Step 1 — Load and profile the data

import pandas as pd
import json

# CSV
df = pd.read_csv("/mnt/user-data/uploads/data.csv")
print(f"Shape: {df.shape}")
print(f"Columns: {df.columns.tolist()}")
print(f"Dtypes:\n{df.dtypes}")
print(f"Null counts:\n{df.isnull().sum()}")

Step 2 — Column-level validation

import re
from datetime import datetime

errors = []  # collect all errors

def log_error(row_idx, column, value, reason):
    errors.append({
        "row": row_idx + 2,  # +2 for 1-indexed + header row
        "column": column,
        "value": str(value)[:50],
        "error": reason
    })

# --- Null / required field check ---
required_cols = ["id", "email", "name"]
for col in required_cols:
    nulls = df[df[col].isnull()]
    for idx in nulls.index:
        log_error(idx, col, None, "Required field is null")

# --- Email format ---
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
for idx, row in df.iterrows():
    if pd.notna(row.get("email")) and not EMAIL_RE.match(str(row["email"])):
        log_error(idx, "email", row["email"], "Invalid email format")

# --- Phone number (E.164 format) ---
PHONE_RE = re.compile(r"^\+?[1-9]\d{7,14}$")
for idx, row in df.iterrows():
    phone = str(row.get("phone", "")).replace(" ", "").replace("-", "")
    if phone and not PHONE_RE.match(phone):
        log_error(idx, "phone", row["phone"], "Invalid phone format")

# --- Numeric range ---
for idx, row in df.iterrows():
    age = row.get("age")
    if pd.notna(age):
        if not (0 < int(age) < 130):
            log_error(idx, "age", age, "Age out of valid range (0–130)")

# --- Date format ---
DATE_FORMAT = "%Y-%m-%d"
for idx, row in df.iterrows():
    d = row.get("date")
    if pd.notna(d):
        try:
            datetime.strptime(str(d), DATE_FORMAT)
        except ValueError:
            log_error(idx, "date", d, f"Invalid date — expected {DATE_FORMAT}")

# --- Categorical / allowed values ---
ALLOWED_STATUS = {"active", "inactive", "pending"}
for idx, row in df.iterrows():
    status = row.get("status")
    if pd.notna(status) and str(status).lower() not in ALLOWED_STATUS:
        log_error(idx, "status", status, f"Invalid status — must be one of {ALLOWED_STATUS}")

# --- Duplicate IDs ---
dupes = df[df.duplicated(subset=["id"], keep=False)]
for idx in dupes.index:
    log_error(idx, "id", df.loc[idx, "id"], "Duplicate ID")

Step 3 — Generate validation report

error_df = pd.DataFrame(errors)

if error_df.empty:
    print("✅ All validation checks passed. No errors found.")
else:
    print(f"❌ Found {len(error_df)} validation errors across {error_df['row'].nunique()} rows\n")

    # Summary by error type
    print("Error summary:")
    for err_type, count in error_df["error"].value_counts().items():
        print(f"  {count:4d}x  {err_type}")

    print(f"\nFirst 20 errors:")
    print(error_df.head(20).to_string(index=False))

    # Save full report
    error_df.to_csv("/mnt/user-data/outputs/validation_report.csv", index=False)

    # Save clean rows (passed validation)
    bad_rows = set(error_df["row"] - 2)  # convert back to df index
    clean_df = df.drop(index=[i for i in bad_rows if i in df.index])
    clean_df.to_csv("/mnt/user-data/outputs/data_clean.csv", index=False)
    print(f"\nClean rows: {len(clean_df)}/{len(df)} saved to data_clean.csv")

Step 4 — JSON schema validation

import jsonschema, json

schema = {
    "type": "array",
    "items": {
        "type": "object",
        "required": ["id", "name", "email"],
        "additionalProperties": False,
        "properties": {
            "id": {"type": "integer", "minimum": 1},
            "name": {"type": "string", "minLength": 1},
            "email": {"type": "string", "format": "email"},
            "age": {"type": "integer", "minimum": 0, "maximum": 130},
            "status": {"type": "string", "enum": ["active", "inactive"]}
        }
    }
}

with open("/mnt/user-data/uploads/data.json") as f:
    data = json.load(f)

validator = jsonschema.Draft7Validator(schema)
errors = list(validator.iter_errors(data))

if not errors:
    print("✅ JSON is valid")
else:
    print(f"❌ {len(errors)} validation errors:")
    for e in errors[:10]:
        path = " → ".join(str(p) for p in e.path)
        print(f"  [{path}] {e.message}")

Edge cases

| Situation | Fix | |-----------|-----| | Mixed types in column | Cast to string and validate format with regex | | Dates in various formats | Use pd.to_datetime(errors='coerce') then flag NaT | | Unicode in text fields | Check with str.isprintable() if ASCII-only required | | Large file (1M+ rows) | Validate in chunks: pd.read_csv(f, chunksize=10000) | | Referential integrity | Join to reference table and flag rows with no match | | Business rule validation | Add custom validate_* functions per rule |

Output format

Always produce:

1. Summary: total rows, rows with errors, rows clean
2. Error breakdown by type (count per error category)
3. Sample of first 20 errors with row number, column, value, reason
4. Two output files: validation_report.csv and data_clean.csv

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: prasad-nimbalkar
• Source: prasad-nimbalkar/claude-agent-skills
• License: MIT

Install and usage instructions live in the source repository linked above.