Request Validation Pipeline — Mini Project
In this tutorial, you'll build a complete Request Validation Pipeline. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
This mini project combines schema validation, sanitization, type coercion, custom validators, async uniqueness checks, and standardized error handling into a single pipeline.
What You'll Learn
By the end of this lesson, you will build a production-ready validation pipeline that handles body validation, query validation, sanitization, async checks, and returns structured errors.
Why It Matters
A well-designed validation pipeline is the single most important security control for your API. This project gives you a reusable foundation for any Express or Python web framework.
Real-World Use
This pattern mirrors how Stripe, GitHub, and Twilio validate API requests: schema validation, business rule checks, and async database lookups — all before business logic.
Project Architecture
flowchart TD
Request --> BodyParser[Body Parser]
BodyParser --> TypeCoerce[Type Coercion]
TypeCoerce --> SchemaValidate[Schema Validation]
SchemaValidate --> Sanitize[Input Sanitization]
Sanitize --> CustomValidators[Custom Validators]
CustomValidators --> AsyncChecks[Async DB Checks]
AsyncChecks --> Handler[Route Handler]
CustomValidators --> ErrorHandler[Error Handler]
SchemaValidate --> ErrorHandler
Project Structure
validation_pipeline/
schema.py # Schema definitions and validators
sanitizer.py # Input sanitization
coercer.py # Type coercion
validators.py # Custom and async validators
pipeline.py # Pipeline orchestrator
errors.py # Error response builder
app.py # Main application
Schema and Coercion
# schema.py
from typing import Any, Dict, List, Optional, Type
class FieldSchema:
def __init__(self, field_type: Type, required: bool = False,
min_length: Optional[int] = None, max_length: Optional[int] = None,
min_value: Optional[float] = None, max_value: Optional[float] = None,
pattern: Optional[str] = None, enum: Optional[List] = None):
self.field_type = field_type
self.required = required
self.min_length = min_length
self.max_length = max_length
self.min_value = min_value
self.max_value = max_value
self.pattern = pattern
self.enum = enum
class Schema:
def __init__(self):
self.fields: Dict[str, FieldSchema] = {}
def add(self, name: str, **kwargs):
self.fields[name] = FieldSchema(**kwargs)
def coerce_and_validate(self, data: Dict) -> tuple:
coerced = {}
errors = []
for name, schema in self.fields.items():
value = data.get(name)
if schema.required and value is None:
errors.append({"field": name, "code": "required"})
continue
if value is None:
continue
coerced_value = self._coerce(value, schema.field_type)
if coerced_value is None and value is not None:
errors.append({"field": name, "code": "type_error",
"expected": schema.field_type.__name__})
continue
validation_error = self._validate_field(name, coerced_value, schema)
if validation_error:
errors.append(validation_error)
else:
coerced[name] = coerced_value
return coerced, errors
def _coerce(self, value: Any, target: Type) -> Any:
if isinstance(value, target):
return value
try:
return target(value)
except (ValueError, TypeError):
return None
def _validate_field(self, name: str, value: Any, schema: FieldSchema) -> Optional[Dict]:
errors = []
if schema.field_type == str and isinstance(value, str):
if schema.min_length and len(value) < schema.min_length:
errors.append({"field": name, "code": "min_length", "min": schema.min_length})
if schema.max_length and len(value) > schema.max_length:
errors.append({"field": name, "code": "max_length", "max": schema.max_length})
if schema.pattern:
import re
if not re.match(schema.pattern, value):
errors.append({"field": name, "code": "pattern"})
if schema.field_type in (int, float):
if schema.min_value is not None and value < schema.min_value:
errors.append({"field": name, "code": "min_value", "min": schema.min_value})
if schema.max_value is not None and value > schema.max_value:
errors.append({"field": name, "code": "max_value", "max": schema.max_value})
if schema.enum and value not in schema.enum:
errors.append({"field": name, "code": "invalid_enum", "valid": schema.enum})
return errors[0] if errors else None
user_schema = Schema()
user_schema.add("username", field_type=str, required=True, min_length=3, max_length=30, pattern=r"^[a-zA-Z0-9_]+$")
user_schema.add("email", field_type=str, required=True)
user_schema.add("age", field_type=int, required=True, min_value=13, max_value=150)
user_schema.add("role", field_type=str, required=True, enum=["admin", "user"])
coerced, errors = user_schema.coerce_and_validate({
"username": "alice_123", "email": "a@x.com", "age": "30", "role": "admin"
})
print(f"Valid: coercd={coerced}, errors={errors}")
coerced2, errors2 = user_schema.coerce_and_validate({
"username": "a", "email": "a@x.com", "age": 0, "role": "superadmin"
})
print(f"Invalid: errors={errors2}")
Expected output:
Valid: coercd={'username': 'alice_123', 'email': 'a@x.com', 'age': 30, 'role': 'admin'}, errors=[]
Invalid: errors=[{'field': 'username', 'code': 'min_length', 'min': 3}, {'field': 'age', 'code': 'min_value', 'min': 13}, {'field': 'role', 'code': 'invalid_enum', 'valid': ['admin', 'user']}]
Pipeline Orchestrator
# pipeline.py
import asyncio
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))
async def run(self, data: Dict) -> Dict:
context = {
"data": data,
"coerced": {},
"errors": [],
"sanitized": {},
"warnings": [],
}
for step in self.steps:
if context["errors"]:
break
try:
result = step.handler(context)
if asyncio.iscoroutine(result):
result = await result
if result is not None:
context.update(result)
except Exception as e:
context["errors"].append({
"step": step.name,
"code": "internal_error",
"message": str(e),
})
break
context["valid"] = len(context["errors"]) == 0
return context
pipeline = ValidationPipeline()
def parse_and_coerce(ctx):
from schema import user_schema
coerced, errors = user_schema.coerce_and_validate(ctx["data"])
return {"coerced": coerced, "errors": ctx["errors"] + errors}
def sanitize(ctx):
result = {}
for k, v in ctx["coerced"].items():
if isinstance(v, str):
result[k] = v.strip()
else:
result[k] = v
return {"sanitized": result}
async def check_uniqueness(ctx):
await asyncio.sleep(0.01)
taken = {"alice@x.com"}
email = ctx["sanitized"].get("email", "")
if email in taken:
ctx["errors"].append({"field": "email", "code": "not_unique"})
pipeline.add("coerce", parse_and_coerce)
pipeline.add("sanitizer", sanitize)
pipeline.add("uniqueness", check_uniqueness)
async def main():
result = await pipeline.run({
"username": "new_user", "email": "new@x.com", "age": "25", "role": "user"
})
print(f"New user: valid={result['valid']}, errors={result['errors']}")
result2 = await pipeline.run({
"username": "new_user", "email": "alice@x.com", "age": "25", "role": "user"
})
print(f"Taken email: valid={result2['valid']}, errors={result2['errors']}")
asyncio.run(main())
Expected output:
New user: valid=True, errors=[]
Taken email: valid=False, errors=[{'field': 'email', 'code': 'not_unique'}]
Error Response Builder
# errors.py
from typing import Any, Dict, List
class ErrorResponseBuilder:
@staticmethod
def build(errors: List[Dict]) -> Dict:
return {
"status": 400,
"error": "Validation failed",
"details": [
{
"field": e.get("field", "unknown"),
"code": e.get("code", "invalid"),
}
for e in errors
],
}
@staticmethod
def field_error(field: str, code: str) -> Dict:
return {"field": field, "code": code}
builder = ErrorResponseBuilder()
errors = [
{"field": "email", "code": "required"},
{"field": "age", "code": "min_value"},
]
print(builder.build(errors))
Expected output:
{'status': 400, 'error': 'Validation failed', 'details': [{'field': 'email', 'code': 'required'}, {'field': 'age', 'code': 'min_value'}]}
Common Mistakes
1. Inconsistent Coercion
Coercing some fields but not others. The pipeline must ensure all fields that pass validation are properly typed.
2. Sync Blocking in Async Pipeline
Database lookups in synchronous steps block the event loop. Use async steps for I/O.
3. No Rollback
If validation passes but business logic fails, the database may have stale data. Transaction rollback is needed.
4. Skipping Sanitization
Validated data can still contain XSS payloads. Sanitize after validation.
5. Error Leakage
Including internal details like SQL queries in error messages. Always format errors with safe messages.
Practice Questions
1. What is the correct order of pipeline steps?
Parse -> Coerce -> Validate -> Sanitize -> Custom/Async checks -> Handler.
2. Why coerce before validate?
Validation checks types. Without coercion, all HTTP string input fails type checks.
3. How do you handle async validation failures?
Return validation errors just like sync failures. The pipeline should collect all errors regardless of type.
4. What makes a validation pipeline production-ready?
Schema validation, type coercion, sanitization, custom rules, async checks, standardized errors, and global error handling.
Challenge
Extend the pipeline with: query parameter validation, header validation, file upload validation (type + size), rate limit checking, and request logging.
FAQ
What's Next
You completed the validation pipeline project. Next, explore data validation, then learn about file upload handling.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro