data-validation-helper

Data Validation Helper

Workflow en 8 étapes

1. Inventaire et règles métier

• Lister chaque colonne : type attendu, nullabilité, plage (min/max), format (regex, pattern date), unicité, enum de valeurs autorisées.
• Documenter dans un fichier versionné (expectations.yaml, schema.yaml ou conftest.py).
• Critère clé : distinguer règle bloquante (type incorrect, PK dupliquée) de warning tolerable (outlier statistique, valeur rare).

2. Schema validation — fail fast

JSON Schema (API REST) :

from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "user_id": {"type": "integer"},
        "email":   {"type": "string", "format": "email"},
        "amount":  {"type": "number",  "minimum": 0}
    },
    "required": ["user_id", "email", "amount"]
}

try:
    validate(instance=payload, schema=schema)
except ValidationError as e:
    raise ValueError(f"Payload invalide : {e.message}")

Pydantic v2 (Python objects) :

from pydantic import BaseModel, field_validator
from decimal import Decimal

class Transaction(BaseModel):
    user_id: int
    amount: Decimal
    currency: str

    @field_validator("currency")
    @classmethod
    def check_currency(cls, v: str) -> str:
        if v not in {"TND", "EUR", "USD"}:
            raise ValueError(f"Devise inconnue : {v}")
        return v

Pandera (DataFrames) :

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int,  nullable=False),
    "amount":  pa.Column(float, pa.Check.ge(0)),
    "status":  pa.Column(str,  pa.Check.isin(["pending","done","failed"]))
})

validated_df = schema.validate(df, lazy=True)  # lazy=True collecte toutes les erreurs

3. Dimensions de qualité à mesurer