data-validation

name	data-validation
description	Validate data with schemas across languages and formats. Use when defining JSON Schema, using Zod (TypeScript) or Pydantic (Python), validating API request/response shapes, checking CSV/JSON data integrity, or setting up data contracts between services.
metadata	{"clawdbot":{"emoji":"✅","requires":{"anyBins":["node","python3","jq"]},"os":["linux","darwin","win32"]}}

Data Validation

Schema-based data validation across languages and formats. Covers JSON Schema, Zod (TypeScript), Pydantic (Python), API boundary validation, data contracts, and integrity checking.

When to Use

• Defining the shape of API request/response bodies
• Validating user input before processing
• Setting up data contracts between services
• Checking CSV/JSON file integrity before import
• Migrating data (did the ETL preserve everything?)
• Generating types or documentation from schemas

JSON Schema

Basic schema

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "required": ["name", "email", "age"],
  "properties": {
    "name": {
      "type": "string",
      "minLength": 1,
      "maxLength": 100
    },
    "email": {
      "type": "string",
      "format": "email"
    },
    "age": {
      "type": "integer",
      "minimum": 0,
      "maximum": 150
    },
    "role": {
      "type": "string",
      "enum": ["user", "admin", "moderator"],
      "default": "user"
    },
    "tags": {
      "type": "array",
      "items": { "type": "string" },
      "uniqueItems": true,
      "maxItems": 10
    },
    "address": {
      "type": "object",
      "properties": {
        "street": { "type": "string" },
        "city": { "type": "string" },
        "zip": { "type": "string", "pattern": "^\\d{5}(-\\d{4})?$" }
      },
      "required": ["street", "city"]
    }
  },
  "additionalProperties": false
}

Common patterns

// Nullable field
{ "type": ["string", "null"] }

// Union type (string or number)
{ "oneOf": [{ "type": "string" }, { "type": "number" }] }

// Conditional: if role is admin, require permissions
{
  "if": { "properties": { "role": { "const": "admin" } } },
  "then": { "required": ["permissions"] }
}

// Pattern properties (dynamic keys)

   
   
        
  




   
     
       
       
            
            
      
    
  
   
        
        
  

Validate with command line

# Using ajv-cli (Node.js)
npx ajv-cli validate -s schema.json -d data.json

# Using jsonschema (Python)
pip install jsonschema
python3 -c "
import json, jsonschema
schema = json.load(open('schema.json'))
data = json.load(open('data.json'))
jsonschema.validate(data, schema)
print('Valid')
"

# Validate multiple files
for f in data/*.json; do
  npx ajv-cli validate -s schema.json -d "$f" 2>&1 || echo "INVALID: $f"
done

Zod (TypeScript)

Basic schemas

import { z } from 'zod';

// Primitives
const nameSchema = z.string().min(1).max(100);
const ageSchema = z.number().int().min(0).max(150);
const emailSchema = z.string().email();
const urlSchema = z.string().url();

// Objects
const userSchema = z.object({
  name: z.string().min(1),
  email: z.string().email(),
  age: z.number().int().min(0),
  role: z.enum(['user', 'admin', 'moderator']).default('user'),
  tags: z.array(z.string()).max(10).default([]),
  createdAt: z.().(),
});


  = z.< userSchema>;



 result = userSchema.(data);
 (result.) {
  .(result.); 
}  {
  .(result..); 
}


 user = userSchema.(data);

Advanced patterns

// Optional and nullable
const schema = z.object({
  name: z.string(),
  nickname: z.string().optional(),       // string | undefined
  middleName: z.string().nullable(),     // string | null
  suffix: z.string().nullish(),          // string | null | undefined
});

// Transforms (validate then transform)
const dateSchema = z.string().datetime().transform(s => new Date(s));
const trimmed = z.string().trim().toLowerCase();
const parsed = z.string().transform(s => parseInt(s, 10)).pipe(z.number().int());

// Discriminated unions (tagged unions)
const eventSchema = z.discriminatedUnion('type', [
  z.object({ type: z.literal('click'), x: z.(), : z.() }),
  z.({ : z.(), : z.() }),
  z.({ : z.(), : z.() }),
]);


 : z.<> = z.({
  : z.(),
  : z.( z.(categorySchema)).([]),
});


 passwordSchema = z.()
  .()
  .( .(s), )
  .( .(s), )
  .( .(s), );


 baseUser = z.({ : z.(), : z.() });
 adminUser = baseUser.({ : z.(z.()) });


 createUser = userSchema.({ :  });
 userSummary = userSchema.({ : , :  });


 flexible = userSchema.();


 strict = userSchema.(); 

API validation with Zod

// Express middleware
import { z } from 'zod';

const createUserBody = z.object({
  name: z.string().min(1),
  email: z.string().email(),
  password: z.string().min(8),
});

app.post('/api/users', (req, res) => {
  const result = createUserBody.safeParse(req.body);
  if (!result.success) {
    return res.status(400).json({ errors: result.error.issues });
  }
  const { name, email, password } = result.data;
  // ... create user
});

// Query parameter validation
const listParams = z.object({
  page: z.coerce.number().int().min(1).default(1),
  limit: z.coerce.number().int().().().(),
  : z.([, , ]).(),
  : z.().(),
});

app.(,  {
   params = listParams.(req.);
  
});

Pydantic (Python)

Basic models

from pydantic import BaseModel, Field, EmailStr, field_validator
from typing import Optional
from datetime import datetime
from enum import Enum

class Role(str, Enum):
    USER = "user"
    ADMIN = "admin"
    MODERATOR = "moderator"

class Address(BaseModel):
    street: str
    city: str
    zip_code: str = Field(pattern=r"^\d{5}(-\d{4})?$")

class User(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(ge=0, le=150)
    role: Role = Role.USER
    tags: list[str] = Field(default_factory=list, max_length=10)
    address: Optional[Address] = None
    created_at: datetime = Field(default_factory=datetime.now)

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError()
         v.strip()


user = User(name=, email=, age=)
(user.model_dump())      
(user.model_dump_json())  


:
    User(name=, email=, age=-)
 Exception  e:
    (e)  

Advanced patterns

from pydantic import BaseModel, model_validator, ConfigDict
from typing import Literal, Union, Annotated

# Discriminated union
class ClickEvent(BaseModel):
    type: Literal["click"]
    x: int
    y: int

class KeypressEvent(BaseModel):
    type: Literal["keypress"]
    key: str

Event = Annotated[Union[ClickEvent, KeypressEvent], Field(discriminator="type")]

# Model-level validation (cross-field)
class DateRange(BaseModel):
    start: datetime
    end: datetime

    @model_validator(mode="after")
    def end_after_start(self):
        if self.end <= self.start:
            raise ValueError("end must be after start")
        return self

# Strict mode (no type coercion)
class StrictUser(BaseModel):
    model_config = ConfigDict(strict=True)
    age: int  # "30" will be rejected, must be int 30


 ():
    user_name:  = Field(alias=)
    created_at: datetime = Field(alias=)

    model_config = ConfigDict(populate_by_name=)


 pydantic  computed_field

 ():
    items: []
    tax_rate:  = 



     () -> :
        subtotal = (i.get(, ) * i.get(, )  i  .items)
         (subtotal * ( + .tax_rate), )


(User.model_json_schema())

FastAPI integration

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class CreateUser(BaseModel):
    name: str = Field(min_length=1)
    email: EmailStr
    password: str = Field(min_length=8)

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

@app.post("/api/users", response_model=UserResponse)
async def create_user(body: CreateUser):
    # body is already validated
    return {"id": 1, "name": body.name, "email": body.email}

@app.get("/api/users")
async def list_users(
    page: int = Query(default=1, ge=1),
    limit: int = Query(default=20, ge=1, le=100),
    q: str | None = Query(default=None),
):
    # All params validated and typed
    

Data Integrity Checks

CSV validation

#!/bin/bash
# validate-csv.sh — Check CSV structure and data quality
FILE="${1:?Usage: validate-csv.sh <file.csv>}"

echo "=== CSV Validation: $FILE ==="

# Row count
ROWS=$(wc -l < "$FILE")
echo "Rows: $ROWS (including header)"

# Column count consistency
HEADER_COLS=$(head -1 "$FILE" | awk -F',' '{print NF}')
echo "Columns (header): $HEADER_COLS"

BAD_ROWS=$(awk -F',' -v expected="$HEADER_COLS" 'NR>1 && NF!=expected {count++} END {print count+0}' "$FILE")
if [ "$BAD_ROWS" -gt 0 ]; then
    echo "ERROR: $BAD_ROWS rows have wrong column count"
    awk -F',' -v expected="$HEADER_COLS" 'NR>1 && NF!=expected {print "  Line "NR": "NF" columns (expected "expected")"}' "$FILE" | head -5
else
    echo "Column count: consistent"
fi

# Empty fields
EMPTY=$(awk -F',' '{for(i=1;i<=NF;i++) if($i=="") count++} END {print count}' )
 


DUPES=$(($(sort "" | uniq -d | wc -l)))
 

 

JSON validation

# Check if file is valid JSON
jq empty data.json && echo "Valid JSON" || echo "Invalid JSON"

# Validate structure of each object in an array
jq -e '
  .[] |
  select(
    (.name | type) != "string" or
    (.email | type) != "string" or
    (.age | type) != "number" or
    .age < 0
  )
' data.json && echo "INVALID records found" || echo "All records valid"

# Check for required fields
jq -e '.[] | select(.id == null or .name == null)' data.json

# Check for unique IDs
jq '[.[].id] | length != (. | unique | length)' data.json
# true = duplicates exist

# Compare record counts between source and target
SRC=$(jq length source.json)
TGT=$(jq length target.json)
echo "Source: $SRC, Target: $TGT, Match: $([ "$SRC" = "$TGT" ] && echo yes || echo NO)"

Migration validation

#!/usr/bin/env python3
"""Validate that a data migration preserved all records."""
import json
import sys

def validate_migration(source_path, target_path, key_field="id"):
    with open(source_path) as f:
        source = {r[key_field]: r for r in json.load(f)}
    with open(target_path) as f:
        target = {r[key_field]: r for r in json.load(f)}

    missing = set(source) - set(target)
    extra = set(target) - set(source)
    changed = []

    for key in set(source) & set(target):
        if source[key] != target[key]:
            changed.append(key)

    print(f"Source records: {len(source)}")
    print(f"Target records: {len(target)}")
    print(f"Missing in target: {len(missing)}")
    print(f"Extra in target: {len(extra)}")
    print(f"Changed: {len(changed)}")

    if missing:
        print()
     extra:
        ()
     changed:
        ()
         key  changed[:]:
            ()
             field  (source[key]) | (target[key]):
                s = source[key].get(field)
                t = target[key].get(field)
                 s != t:
                    ()

     (missing) ==   (extra) == 

 __name__ == :
    ok = validate_migration(sys.argv[], sys.argv[], sys.argv[]  (sys.argv) >   )
    sys.exit(  ok  )

Tips

• Validate at system boundaries (API endpoints, file imports, message queues), not deep inside business logic. Trust internal data.
• Zod and Pydantic both generate JSON Schema from their definitions. Use this for documentation, OpenAPI specs, and cross-language contracts.
• additionalProperties: false in JSON Schema catches typos in field names. Use it for strict APIs.
• Pydantic v2 is significantly faster than v1. Use model_config = ConfigDict(strict=True) when you want no implicit type coercion.
• Zod's .safeParse() returns a result object; .parse() throws. Use safeParse in API handlers to return structured errors.
• For CSV validation, always check column count consistency first — most downstream errors trace back to misaligned columns.
• Data migration validation should compare record counts, check for missing/extra records, and sample-check field values. Counting alone isn't enough.