Introduction to Validation Pipelines
In this tutorial, you'll learn about Validation Pipelines. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
A request validation pipeline is a middleware chain that validates, sanitizes, and enriches incoming HTTP requests before they reach your business logic.
What You'll Learn
By the end of this lesson, you will understand validation pipeline architecture, implement a basic pipeline, and know how validation prevents security vulnerabilities.
Why It Matters
Without validation, your API accepts any input. Invalid data causes crashes, security holes, and data corruption. A validation pipeline is your first line of defense.
Real-World Use
Express.js applications commonly use Joi or Zod validation middleware. Every request passes through validation before the route handler processes it.
Pipeline Architecture
flowchart LR
Request[Incoming Request] --> Parse[Parse Body/Params]
Parse --> Validate[Schema Validation]
Validate --> Sanitize[Input Sanitization]
Sanitize --> Coerce[Type Coercion]
Coerce --> Custom[Custom Validators]
Custom --> Handler[Business Logic]
Custom --> Error[Error Handler]
Error --> Response[Error Response]
Basic Pipeline Implementation
# basic_pipeline.py
from typing import Any, Callable, Dict, List, Optional
class PipelineStep:
def __init__(self, name: str, handler: Callable):
self.name = name
self.handler = handler
class ValidationPipeline:
def __init__(self):
self.steps: List[PipelineStep] = []
def add(self, name: str, handler: Callable):
self.steps.append(PipelineStep(name, handler))
def execute(self, request: Dict) -> Dict:
context = {"request": request, "errors": [], "sanitized": {}}
for step in self.steps:
try:
result = step.handler(context)
if result is not None:
context = result
except Exception as e:
context["errors"].append({"step": step.name, "error": str(e)})
break
return context
pipeline = ValidationPipeline()
pipeline.add("parse_body", lambda ctx: {**ctx, "parsed": ctx["request"]})
pipeline.add("validate_email", lambda ctx: {
**ctx, "errors": ctx["errors"] + (["Invalid email"]
if "@" not in ctx["parsed"].get("email", "") else [])
})
pipeline.add("sanitize", lambda ctx: {
**ctx, "sanitized": {k: str(v).strip() for k, v in ctx["parsed"].items()}
})
valid_request = {"email": "user@example.com", "name": " Alice "}
result = pipeline.execute(valid_request)
print(f"Valid: errors={result['errors']}, sanitized={result['sanitized']}")
invalid_request = {"email": "invalid", "name": "Bob"}
result2 = pipeline.execute(invalid_request)
print(f"Invalid: errors={result2['errors']}")
Expected output:
Valid: errors=[], sanitized={'email': 'user@example.com', 'name': 'Alice'}
Invalid: errors=[{'step': 'validate_email', 'error': 'Invalid email'}]
Pipeline with Validation Result
# pipeline_result.py
from typing import Any, Dict, List, Optional
class ValidationResult:
def __init__(self):
self.passed = True
self.errors: List[Dict] = []
self.data: Dict = {}
self.warnings: List[str] = []
def fail(self, field: str, message: str):
self.passed = False
self.errors.append({"field": field, "message": message})
def warn(self, message: str):
self.warnings.append(message)
def merge(self, data: Dict):
self.data.update(data)
def to_dict(self) -> Dict:
return {
"valid": self.passed,
"errors": self.errors,
"data": self.data,
"warnings": self.warnings,
}
class RequestValidator:
def __init__(self):
self.rules: Dict[str, List[Callable]] = {}
def rule(self, field: str, validator: Callable):
self.rules.setdefault(field, []).append(validator)
def validate(self, data: Dict) -> ValidationResult:
result = ValidationResult()
for field, validators in self.rules.items():
value = data.get(field)
for validator in validators:
error = validator(field, value, data)
if error:
result.fail(field, error)
if result.passed:
result.merge(data)
return result
validator = RequestValidator()
def required(field, value, data):
return f"{field} is required" if value is None or value == "" else None
def min_length(n):
def check(field, value, data):
if value and len(value) < n:
return f"{field} must be at least {n} chars"
return None
return check
validator.rule("name", required)
validator.rule("name", min_length(3))
validator.rule("email", required)
result = validator.validate({"name": "Al", "email": ""})
print(result.to_dict())
result2 = validator.validate({"name": "Alice", "email": "a@x.com"})
print(result2.to_dict())
Expected output:
{'valid': False, 'errors': [{'field': 'name', 'message': 'name must be at least 3 chars'}, {'field': 'email', 'message': 'email is required'}], 'data': {}, 'warnings': []}
{'valid': True, 'errors': [], 'data': {'name': 'Alice', 'email': 'a@x.com'}, 'warnings': []}
Common Mistakes
1. Validating Only at the Controller
Validation must happen at the boundary, not in business logic. Validate early, validate once.
2. No Sanitization
Validating input is not enough. Sanitize strings, strip whitespace, remove dangerous characters.
3. Stopping at First Error
Report all validation errors at once so clients can fix everything in one request.
4. Mixing Validation with Business Logic
Validation belongs in middleware. Business logic in handlers. Keep them separate.
5. No Pipeline Ordering
Parse first, validate second, sanitize third. Wrong order misses issues or double-processes data.
Practice Questions
1. What is a validation pipeline?
A chain of middleware steps that parse, validate, sanitize, and enrich requests before business logic.
2. Why validate in middleware rather than handlers?
Separation Of Concerns. Handlers focus on business logic. Validation is a cross-cutting concern.
3. What is the first step in a validation pipeline?
Parsing the request body/params/query into a structured format.
4. Should validation stop on first error?
No. Return all errors so clients can fix everything at once.
Challenge
Build a validation pipeline for a user registration endpoint with these steps: parse, validate required fields, sanitize strings, check email format, check password strength, and return all errors at once.
FAQ
Mini Project: Pipeline Builder
# pipeline_builder.py
from typing import Any, Callable, Dict, List
class PipelineBuilder:
def __init__(self):
self.steps: List[Callable] = []
def use(self, step: Callable):
self.steps.append(step)
return self
def run(self, data: Dict) -> Dict:
ctx = {"data": data, "errors": [], "passed": True}
for step in self.steps:
ctx = step(ctx)
if ctx.get("errors"):
ctx["passed"] = False
break
return ctx
pipeline = PipelineBuilder()
pipeline.use(lambda ctx: {**ctx, "data": {k: v.strip() if isinstance(v, str) else v for k, v in ctx["data"].items()}})
pipeline.use(lambda ctx: {**ctx, "errors": ctx["errors"] + (["name required"] if not ctx["data"].get("name") else [])})
print(pipeline.run({"name": " Alice "}))
print(pipeline.run({"name": ""}))
Expected output:
{'data': {'name': 'Alice'}, 'errors': [], 'passed': True}
{'data': {'name': ''}, 'errors': ['name required'], 'passed': False}
What's Next
You understand validation pipeline basics. Next, learn why validation matters, then schema validation.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro