loom-data-validation

name	loom-data-validation
description	Data validation patterns covering schema validation, input sanitization, output encoding, and type coercion. Use for form/API validation with Zod/Pydantic/Joi/JSON Schema, XSS and injection prevention, constraint checks, data pipeline and ML feature validation.
triggers	["validate","validation","schema","form validation","API validation","JSON Schema","Zod","Pydantic","Joi","Yup","Ajv","class-validator","sanitize","sanitization","XSS prevention","injection prevention","escape","encode","whitelist","blacklist","constraint checking","invariant validation","data pipeline validation","ML feature validation","custom validators","Great Expectations","data quality","data drift"]

Data Validation

Overview

Data validation ensures that input data meets expected formats, types, and constraints before processing. This skill covers schema validation libraries, input sanitization, output encoding, type coercion strategies, security-focused validation (XSS, injection prevention), data pipeline validation, and comprehensive error handling.

Trigger Keywords

Use this skill when working with:

• Schema validation: JSON Schema, Zod, Pydantic, Joi, Yup, Ajv, class-validator
• Input processing: validate, validation, sanitize, sanitization, input validation, form validation
• Security validation: XSS prevention, injection prevention, escape, encode, whitelist, blacklist
• Constraints: constraint checking, invariant validation, business rules, data quality
• API validation: request validation, response validation, API contracts
• Data pipelines: Great Expectations, dbt tests, data quality checks
• ML/AI: feature validation, distribution checks, data drift detection

Agent Assignments

Agent	Responsibility
senior-software-engineer (Opus)	DEFAULT. Schema architecture, validation strategy, implementation, XSS/injection prevention, sanitization, infrastructure config validation, pipeline validation, data quality checks
software-engineer (Sonnet)	ONLY for unit tests or boilerplate validators following established patterns

Key Concepts

JSON Schema Validation

import Ajv, { JSONSchemaType, ValidateFunction } from "ajv";
import addFormats from "ajv-formats";

// Initialize Ajv with formats
const ajv = new Ajv({
  allErrors: true, // Return all errors, not just first
  removeAdditional: true, // Remove properties not in schema
  useDefaults: true, // Apply default values
  coerceTypes: true, // Coerce types when possible
});
addFormats(ajv);

// Define schema with TypeScript type
interface CreateUserRequest {
  email: string;
  password: string;
  name: string;
  age?: number;
  role: "user" | "admin" | "moderator";
  preferences?: {
    newsletter: boolean;
    theme: "light" | "dark";
  };
}

const createUserSchema: JSONSchemaType<CreateUserRequest> = {
  : ,
  : {
    : { : , : , :  },
    : {
      : ,
      : ,
      : ,
      :
        ,
    },
    : { : , : , :  },
    : { : , : , : , :  },
    : { : , : [, , ] },
    : {
      : ,
      : {
        : { : , :  },
        : { : , : [, ], :  },
      },
      : [, ],
      : ,
      : ,
    },
  },
  : [, , , ],
  : ,
};


 validateCreateUser = ajv.(createUserSchema);


 validate<T>(
  : <T>,
  : ,
): { : ; : T } | { : ; : [] } {
   ((data)) {
     { : , data };
  }

   : [] = (validator. || []).( ({
    :
      err..(, ).(, ) ||
      err..,
    : (err),
    : err.,
  }));

   { : , errors };
}

 ():  {
   (error.) {
     :
       ;
     :
       ;
     :
       ;
     :
       ;
     :
       ;
     :
       ;
     :
       ;
     :
       ;
    :
       error. || ;
  }
}

Zod Validation (TypeScript)

import { z, ZodError, ZodSchema } from "zod";

// Basic schemas
const emailSchema = z.string().email().max(255);
const passwordSchema = z
  .string()
  .min(12, "Password must be at least 12 characters")
  .max(128)
  .regex(/[a-z]/, "Password must contain a lowercase letter")
  .regex(/[A-Z]/, "Password must contain an uppercase letter")
  .regex(/[0-9]/, "Password must contain a number")
  .regex(/[^a-zA-Z0-9]/, "Password must contain a special character");

// Complex schema with transforms and refinements
const createUserSchema = z
  .object({
    email: emailSchema.transform((e) => e.toLowerCase().trim()),
    password: passwordSchema,
    confirmPassword: z.string(),
    name: z
      .string()
      .min(1)
      .max(100)
      .( n.()),
    : z.().().().().(),
    : z.([, , ]).(),
    : z.(z.().()).().([]),
    : z.(z.(), z.()).(),
    : z
      .({
        : z.().(),
        : z.([, ]).(),
        : z
          .({
            : z.().(),
            : z.().(),
            : z.().(),
          })
          .({}),
      })
      .({}),
  })
  .( data. === data., {
    : ,
    : [],
  })
  .( data); 


  = z.< createUserSchema>;
  = z.< createUserSchema>;


 <T> {
  : ;
  ?: T;
  ?: <{
    : ;
    : ;
  }>;
}

 validateWithZod<T>(
  : <T>,
  : ,
): <T> {
   result = schema.(data);

   (result.) {
     { : , : result. };
  }

   errors = result...( ({
    : err..(),
    : err.,
  }));

   { : , errors };
}


 uniqueEmailSchema = emailSchema.(
   (email) => {
     exists =  db..(email);
     !exists;
  },
  { :  },
);


 formSchema = z.(, [
  z.({
    : z.(),
    : z.().(),
    : z.().(),
    : z.().(),
  }),
  z.({
    : z.(),
    : z.().(),
    : z.().(),
  }),
]);


  {
  : ;
  ?: [];
}

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

Pydantic Validation (Python)

from datetime import datetime
from typing import Optional, List, Literal
from pydantic import (
    BaseModel,
    Field,
    EmailStr,
    validator,
    root_validator,
    constr,
    conint,
)
import re

# Basic model with field validation
class CreateUserRequest(BaseModel):
    email: EmailStr
    password: constr(min_length=12, max_length=128)
    name: constr(min_length=1, max_length=100)
    age: Optional[conint(ge=13, le=150)] = None
    role: Literal['user', 'admin', 'moderator'] = 'user'
    tags: List[str] = Field(default_factory=list, max_items=10)

    class Config:
        # Strip whitespace from strings
        anystr_strip_whitespace = True
        # Validate on assignment
        validate_assignment = True
        # Use enum values
        use_enum_values = True

    @validator('email')
    def email_lowercase(cls, v):
        return v.lower()

    @validator()
     ():
          re.search(, v):
             ValueError()
          re.search(, v):
             ValueError()
          re.search(, v):
             ValueError()
          re.search(, v):
             ValueError()
         v


     ():
         (v) > :
             ValueError()
         v.strip().lower()


 ():
    street: 
    city: 
    state: constr(min_length=, max_length=)
    zip_code: constr(regex=)
    country:  = 

 ():
    user: CreateUserRequest
    addresses: [Address] = Field(default_factory=, max_items=)
    primary_address_index:  = 


     ():
        addresses = values.get(, [])
        primary_index = values.get(, )

         addresses  primary_index >= (addresses):
             ValueError()

         values


 typing  TypeVar, 

T = TypeVar()

 (BaseModel, [T]):
    success: 
    data: [T] = 
    errors: [[]] = 
    timestamp: datetime = Field(default_factory=datetime.utcnow)


 pydantic  validator
 asyncio

 ():
    email: EmailStr


     ():
        
         app.db  user_exists_sync
         user_exists_sync(v):
             ValueError()
         v


 pydantic  ValidationError
 fastapi  HTTPException

 ():
    :
         model_class(**data)
     ValidationError  e:
        errors = []
         error  e.errors():
            errors.append({
                : .join((loc)  loc  error[]),
                : error[],
                : error[],
            })
         HTTPException(status_code=, detail={: errors})

Input Sanitization

import DOMPurify from "dompurify";
import { JSDOM } from "jsdom";
import validator from "validator";

// Server-side DOMPurify setup
const window = new JSDOM("").window;
const purify = DOMPurify(window);

// HTML sanitization
function sanitizeHtml(dirty: string, options?: DOMPurify.Config): string {
  const defaultOptions: DOMPurify.Config = {
    ALLOWED_TAGS: ["b", "i", "em", "strong", "a", "p", "br", "ul", "ol", "li"],
    ALLOWED_ATTR: ["href", "target", "rel"],
    ALLOW_DATA_ATTR: false,
    ADD_ATTR: ["target"], // Add target="_blank" to links
    : [, , , , ],
    : [, , ],
  };

   purify.(dirty, { ...defaultOptions, ...options });
}


 ():  {
   purify.(dirty, {
    : [
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
    ],
    : [, , , , , ],
    : ,
  });
}


 ():  {
   input
    .(, )
    .(, )
    .(, )
    .(, )
    .(, )
    .(, );
}


 ():  {
   filename
    .(, ) 
    .(, ) 
    .(, ) 
    .(, ); 
}


 ():  {
   path = ();
   resolvedPath = path.(basePath, userPath);

   (!resolvedPath.(path.(basePath))) {
      ();
  }

   resolvedPath;
}


  {
  ?: ;
  ?: ;
  ?: ;
  ?: ;
  ?: ;
}

 ():  {
   result = input;

   (options. !== ) {
    result = result.();
  }

   (options.) {
    result = validator.(validator.(result));
  }

   (options.) {
    result = result.();
  }

   (options.) {
    result = result.(
       (, ),
      ,
    );
  }

   (options.) {
    result = result.(, options.);
  }

  
  result = result.(, );

   result;
}


 sanitizers = {
  : 
    (input, {
      : ,
      : ,
      : ,
    }),

  :  validator.(input) || ,

  :  input.(, ).(, ),

  : 
    (input, {
      : ,
      : ,
    })
      .(, )
      .(, ),

  : 
    (input, {
      : ,
      : ,
      : ,
    }),
};

Output Encoding

// HTML encoding
function encodeHtml(str: string): string {
  const entities: Record<string, string> = {
    "&": "&amp;",
    "<": "&lt;",
    ">": "&gt;",
    '"': "&quot;",
    "'": "&#x27;",
    "/": "&#x2F;",
    "`": "&#x60;",
    "=": "&#x3D;",
  };

  return str.replace(/[&<>"'`=/]/g, (char) => entities[char]);
}

// JavaScript string encoding (for embedding in <script> tags)
function encodeJsString(str: string): string {
  return str
    .replace(/\\/g, "\\\\")
    .replace(/'/g, "\\'")
    .replace(/"/g, '\\"')
    .replace(/\n/g, "\\n")
    .replace(, )
    .(, )
    .(, )
    .(, )
    .(, );
}


 ():  {
   (str);
}


 ():  {
   str.(,  {
     hex = char.().();
     ;
  });
}


 ():  {
   .(obj)
    .(, )
    .(, )
    .(, )
    .(, );
}


  =  |  |  |  | ;

 ():  {
   (context) {
     :
       (str);
     :
       (str).(, );
     :
       (str);
     :
       (str);
     :
       (str);
    :
       (str);
  }
}


 ():  {
  
   (str);
}


 ():  {
   strings.( {
     value = values[i - ];
     encoded =
       value ===  ? (value) : (value ?? );
     result + encoded + str;
  });
}


 userInput = ;
 safe = safeHtml;

API Request/Response Validation

// Express middleware for request validation
import { Request, Response, NextFunction } from "express";
import { z, ZodSchema } from "zod";

function validate<T>(
  schema: ZodSchema<T>,
  source: "body" | "query" | "params" = "body",
) {
  return (req: Request, res: Response, next: NextFunction) => {
    const result = schema.safeParse(req[source]);

    if (!result.success) {
      return res.status(422).json({
        error: "Validation Error",
        details: result.error.errors.map((e) => ({
          field: e.path.join("."),
          message: e.message,
        })),
      });
    }

    req[source] = result.data;
    next();
  };
}


 createUserSchema = z.({
  : z.().(),
  : z.().(),
  : z.().().(),
});

app.(, (createUserSchema),  (req, res) => {
  
   user =  (req.);
  res.().(user);
});


 userResponseSchema = z.({
  : z.().(),
  : z.().(),
  : z.(),
  : z.().(),
});

 validateResponse<T>(: <T>, : ): T {
   result = schema.(data);
   (!result.) {
      ();
  }
   result.;
}

Data Pipeline Validation (Great Expectations)

# Great Expectations for data quality validation
import great_expectations as ge
from great_expectations.dataset import PandasDataset

# Load dataset with expectations
df = ge.read_csv('data.csv')

# Basic expectations
df.expect_column_to_exist('user_id')
df.expect_column_values_to_not_be_null('email')
df.expect_column_values_to_be_unique('email')
df.expect_column_values_to_match_regex('email', r'^[^@]+@[^@]+\.[^@]+$')
df.expect_column_values_to_be_in_set('status', ['active', 'inactive', 'pending'])

# Numeric expectations
df.expect_column_values_to_be_between('age', 0, 150)
df.expect_column_mean_to_be_between('price', 10, 1000)

# Date expectations
df.expect_column_values_to_be_dateutil_parseable('created_at')

# Custom expectations
def custom_validation(df):
    # Email domain must match company_domain
    emails = df['email'].str.split('@', expand=True)[1]
    return (emails == df['company_domain']).all()

df.expect_column_pair_values_to_be_equal('email_domain', 'company_domain',
                                          custom_fn=custom_validation)

# Run validation suite
results = df.validate()
if  results[]:
     result  results[]:
          result[]:
            ()



version: 

models:
  - name: users
    columns:
      - name: user_id
        tests:
          - unique
          - not_null
      - name: email
        tests:
          - unique
          - not_null
          - email_format  
      - name: age
        tests:
          - dbt_utils.accepted_range:
              min_value: 
              max_value: 
      - name: status
        tests:
          - accepted_values:
              values: [, , ]
      - name: created_at
        tests:
          - not_null
          - dbt_utils.recency:
              datepart: day
              field: created_at
              interval: 

ML Feature Validation

# Feature validation for ML pipelines
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple

class FeatureValidator:
    def __init__(self, expected_schema: Dict[str, str]):
        self.expected_schema = expected_schema
        self.baseline_stats = {}

    def validate_schema(self, df: pd.DataFrame) -> List[str]:
        errors = []

        # Check column presence
        expected_cols = set(self.expected_schema.keys())
        actual_cols = set(df.columns)

        missing = expected_cols - actual_cols
        if missing:
            errors.append(f"Missing columns: {missing}")

        extra = actual_cols - expected_cols
        if extra:
            errors.append(f"Unexpected columns: {extra}")

        # Check data types
        for col, expected_type in self.expected_schema.items():
            if col in df.columns:
                actual_type = str(df[col].dtype)
                if not actual_type.startswith(expected_type):
                    errors.append()

         errors

     () -> []:
        errors = []

         col  df.select_dtypes(include=[np.number]).columns:
             col   .baseline_stats:
                

            baseline_mean = .baseline_stats[col][]
            baseline_std = .baseline_stats[col][]

            current_mean = df[col].mean()
            current_std = df[col].std()

            
            mean_zscore = ((current_mean - baseline_mean) / baseline_std)
             mean_zscore > threshold:
                errors.append()

            
            variance_ratio = current_std / baseline_std
             variance_ratio <   variance_ratio > :
                errors.append()

         errors

     () -> []:
        errors = []
        null_rates = df.isnull().() / (df)

         col, rate  null_rates.items():
             rate > max_null_rate:
                errors.append()

         errors

     () -> []:
        errors = []

         col, expected  expected_categories.items():
             col   df.columns:
                

            actual = (df[col].dropna().unique())
            expected_set = (expected)

            unexpected = actual - expected_set
             unexpected:
                errors.append()

         errors

     ():
         col  df.select_dtypes(include=[np.number]).columns:
            .baseline_stats[col] = {
                : df[col].mean(),
                : df[col].std(),
                : df[col].(),
                : df[col].(),
            }


validator = FeatureValidator({
    : ,
    : ,
    : ,
    : ,
})


validator.set_baseline(training_df)


errors = []
errors.extend(validator.validate_schema(new_df))
errors.extend(validator.validate_distributions(new_df))
errors.extend(validator.validate_null_rates(new_df))
errors.extend(validator.validate_categorical_values(new_df, {
    : [, , ]
}))

 errors:
     ValueError( + .join(errors))

Infrastructure Configuration Validation

# JSON Schema for Kubernetes config validation
apiVersion: v1
kind: ConfigMap
metadata:
  name: validation-schema
data:
  deployment-schema.json: |
    {
      "$schema": "http://json-schema.org/draft-07/schema#",
      "type": "object",
      "required": ["apiVersion", "kind", "metadata", "spec"],
      "properties": {
        "apiVersion": {
          "type": "string",
          "pattern": "^apps/v1$"
        },
        "kind": {
          "type": "string",
          "enum": ["Deployment"]
        },
        "spec": {
          "type": "object",
          "required": ["replicas", "selector", "template"],
          "properties": {
            "replicas": {
              "type": "integer",
              "minimum": 1,
              "maximum": 100
            },
            "selector": {
              "type": "object",
              "required": ["matchLabels"]
            },
            "template": {
              "type": "object",
              "required": ["metadata", "spec"],
              "properties": {
                "spec": {
                  "type": "object",
                  "required": ["containers"],
                  "properties": {
                    "containers": {
                      "type": "array",
                      "minItems": 1,
                      "items": {
                        "type": "object",
                        "required": ["name", "image"],
                        "properties": {
                          "resources": {
                            "type": "object",
                            "required": ["requests", "limits"]
                          }
                        }
                      }
                    }
                  }
                }
              }
            }
          }
        }
      }
    }

# Terraform configuration validation
import hcl2
import json
from jsonschema import validate, ValidationError

def validate_terraform_config(config_path: str, schema_path: str):
    # Parse HCL
    with open(config_path, 'r') as f:
        config = hcl2.load(f)

    # Load schema
    with open(schema_path, 'r') as f:
        schema = json.load(f)

    # Validate
    try:
        validate(instance=config, schema=schema)
        print("Terraform config is valid")
    except ValidationError as e:
        print(f"Validation error: {e.message}")
        print(f"Path: {' -> '.join(str(p) for p in e.path)}")
        raise

# Custom business rule validation
def validate_aws_resource_tags(config: dict) -> List[str]:
    errors = []
    required_tags = {'Environment', 'Owner', 'CostCenter'}

    for resource in config.get(, {}).values():
         resource_name, resource_config  resource.items():
            tags = (resource_config.get(, {}).keys())
            missing = required_tags - tags

             missing:
                errors.append()

     errors

Best Practices

1. Validate Early

   • Validate at the boundary (API endpoints, form submissions, pipeline ingestion)
   • Fail fast with clear error messages
   • Don't trust any external input

2. Use Schema Validation Libraries

   • Prefer Zod/Pydantic for type safety
   • JSON Schema for language-agnostic validation
   • Generate TypeScript types from schemas

3. Sanitize and Encode

   • Sanitize input based on context (HTML, SQL, paths)
   • Encode output based on where it's rendered
   • Use parameterized queries instead of escaping for SQL

4. Security-First Validation

   • Whitelist allowed values rather than blacklist
   • Prevent XSS with output encoding
   • Prevent injection with parameterized queries and sanitization
   • Validate file uploads (type, size, content)

5. Data Pipeline Validation

   • Validate schema before processing
   • Check data distributions for drift
   • Monitor null rates and cardinality
   • Use Great Expectations for comprehensive data quality

6. ML Feature Validation

   • Validate schema matches training data
   • Detect distribution drift
   • Check for unexpected categories
   • Monitor feature correlations

7. Error Messages

   • Provide specific, actionable error messages
   • Include field names in errors
   • Don't expose internal details in production

8. Defense in Depth

   • Validate on both client and server
   • Apply principle of least privilege
   • Validate at multiple layers (API, service, database)