data-validation

Data Validation Skill

Bad data in → bad decisions out. Validate at every boundary: ingestion, transformation, and output.

1. When to Apply This Skill

Trigger conditions:

• Building or reviewing ETL/ELT pipelines
• "How do we validate incoming data?"
• Designing data contracts between services
• Data quality issues in production
• Schema migration or new data source integration

2. Validation Layers

Validate at three boundaries — not just one.

Source → [Ingestion Validation] → Raw Layer
Raw    → [Transform Validation] → Clean Layer
Clean  → [Output Validation]    → Consumer / API / Report

Layer	What to Check	Fail Strategy
Ingestion	Schema, types, required fields, file format	Reject row or file, log to dead-letter
Transform	Business rules, referential integrity, ranges	Flag row, route to review queue
Output	Final shape matches consumer contract	Block publish, alert team

3. Pydantic Validation (Python)

Schema Definition

from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import date
from enum import Enum

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class ComplaintRecord(BaseModel):
    complaint_id: str = Field(..., min_length=1, max_length=50)
    customer_id: str = Field(..., pattern=r"^CUST-\d{6}$")
    category: str = Field(..., min_length=1)
    priority: Priority
    description: str = Field(..., min_length=10, max_length=5000)
    created_date: date
    resolved_date: date | None = None
    amount: float = Field(default=0, ge=0, le=1_000_000)

    @field_validator("resolved_date")
    @classmethod
    def resolved_after_created(cls, v, info):
        if v and info.data.get("created_date") and v < info.data["created_date"]:
            raise ValueError("Resolved date cannot be before created date")
        return v

    @model_validator(mode="after")
    def critical_must_have_description(self):
        if self.priority == Priority.CRITICAL and len(self.description) < 50:
            raise ValueError("Critical complaints require detailed description (50+ chars)")
        return self