Express Validation Pipeline — Complete Guide
In this tutorial, you'll learn about the Express Validation Pipeline. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Express validation uses middleware functions to validate request data before it reaches route handlers, with schema libraries like Joi or Zod defining the rules.
What You'll Learn
By the end of this lesson, you will build a production-ready Express validation pipeline with schema validation, sanitization, error aggregation, and standardized responses.
Why It Matters
Express is the most popular Node.js framework. Its middleware pattern is ideal for validation pipelines. Well-structured validation prevents bugs and security issues.
Real-World Use
A typical Express production app has 5+ middleware layers: body parser, authentication, validation, sanitization, route handler, and error handler.
Express Validation Architecture
flowchart LR
Request --> BodyParser[Body Parser]
BodyParser --> Auth[Auth Middleware]
Auth --> Validate[Validation Middleware]
Validate --> Sanitize[Sanitization Middleware]
Sanitize --> Handler[Route Handler]
Handler --> Error[Error Handler]
Validate --> Error
Validation Middleware Factory
# express_validation.py
from typing import Any, Callable, Dict, List
class Request:
def __init__(self):
self.body: Dict = {}
self.params: Dict = {}
self.query: Dict = {}
self.validated: Dict = {}
class Response:
def __init__(self):
self.status_code = 200
self.body = {}
def status(self, code: int):
self.status_code = code
return self
def json(self, data: Any):
self.body = data
return self
def validate(schema: Dict[str, Dict]):
def middleware(req: Request, res: Response, next_fn: Callable):
errors = []
for field, rules in schema.items():
value = req.body.get(field)
if rules.get("required") and value is None:
errors.append({"field": field, "code": "required"})
continue
if value is None:
continue
expected_type = rules.get("type")
if expected_type and not isinstance(value, expected_type):
errors.append({"field": field, "code": "type_error", "expected": expected_type.__name__})
continue
min_len = rules.get("minLength")
if min_len and isinstance(value, str) and len(value) < min_len:
errors.append({"field": field, "code": "min_length", "min": min_len})
max_len = rules.get("maxLength")
if max_len and isinstance(value, str) and len(value) > max_len:
errors.append({"field": field, "code": "max_length", "max": max_len})
if errors:
res.status(400).json({"error": "Validation failed", "details": errors})
return
req.validated = {k: req.body.get(k) for k in schema}
next_fn()
return middleware
user_schema = {
"username": {"type": str, "required": True, "minLength": 3, "maxLength": 50},
"email": {"type": str, "required": True},
"age": {"type": int, "required": True},
}
create_user_middleware = validate(user_schema)
def create_user(req: Request, res: Response, next_fn: Callable):
res.status(201).json({"message": "User created", "user": req.validated})
req = Request()
req.body = {"username": "alice", "email": "a@x.com", "age": 30}
res = Response()
create_user_middleware(req, res, lambda: None)
print(f"Valid: status={res.status_code}, body={res.body}")
req2 = Request()
req2.body = {"username": "a", "email": "a@x.com"}
res2 = Response()
create_user_middleware(req2, res2, lambda: None)
print(f"Invalid: status={res2.status_code}, body={res2.body}")
Expected output:
Valid: status=201, body={'message': 'User created', 'user': {'username': 'alice', 'email': 'a@x.com', 'age': 30}}
Invalid: status=400, body={'error': 'Validation failed', 'details': [{'field': 'age', 'code': 'required'}, {'field': 'username', 'code': 'min_length', 'min': 3}]}
Composition with Route-Specific Validation
# route_validation.py
from typing import Any, Callable, Dict, List
class Router:
def __init__(self):
self.routes: Dict = {}
def post(self, path: str, *middleware: Callable):
self.routes[("POST", path)] = list(middleware)
def handle(self, method: str, path: str, req: Request) -> Dict:
handlers = self.routes.get((method, path), [])
res = Response()
chain = iter(handlers)
def next_fn():
try:
handler = next(chain)
handler(req, res, next_fn)
except StopIteration:
pass
next_fn()
return res
router = Router()
def validate_user(req, res, next_fn):
if not req.body.get("name"):
res.status(400).json({"error": "Name required"})
return
req.validated = req.body
next_fn()
def log_request(req, res, next_fn):
print(f" [LOG] {req.body}")
next_fn()
def create_user_handler(req, res, next_fn):
res.status(201).json({"created": req.validated})
router.post("/users", log_request, validate_user, create_user_handler)
req = Request()
req.body = {"name": "Alice"}
res = router.handle("POST", "/users", req)
print(f"Result: status={res.status_code}, body={res.body}")
Expected output:
[LOG] {'name': 'Alice'}
Result: status=201, body={'created': {'name': 'Alice'}}
Error Handling Middleware
# error_handler.py
from typing import Any, Callable, Dict
class ErrorHandler:
@staticmethod
def handle_validation_error(errors: list) -> Dict:
return {
"status": 400,
"error": "Validation failed",
"details": errors,
}
@staticmethod
def handle_type_error(error: TypeError) -> Dict:
return {
"status": 400,
"error": "Type error",
"message": str(error),
}
def wrap(self, fn: Callable) -> Callable:
def wrapper(req, res, next_fn):
try:
fn(req, res, next_fn)
except TypeError as e:
res.status(400).json(self.handle_type_error(e))
except Exception as e:
res.status(500).json({"error": "Internal error"})
return wrapper
handler = ErrorHandler()
safe_handler = handler.wrap(
lambda req, res, n: res.status(200).json({"result": int(req.body.get("value"))})
)
req = Request()
req.body = {"value": "not_a_number"}
res = Response()
safe_handler(req, res, lambda: None)
print(f"Error -> status={res.status_code}, body={res.body}")
Expected output:
Error -> status=400, body={'status': 400, 'error': 'Type error', 'message': "invalid literal for int() with base 10: 'not_a_number'"}
Common Mistakes
1. Validating After Handler
The handler should assume validated data. Validate before the handler runs, not inside it.
2. Not Using next() Properly
Forgetting to call next() blocks the request. Always call next() after successful validation.
3. Inline Validation Logic
Putting validation code inside route handlers instead of reusable middleware.
4. No Global Error Handler
Unhandled validation errors cause 500 responses. A global error handler converts errors to proper responses.
5. Over-Engineering Early
Start with simple middleware validators. Add schema libraries when you need complex rules.
Practice Questions
1. How does Express validation work?
Middleware functions run before the route handler. If validation fails, they return 400. If OK, they call next().
2. What is a validation middleware factory?
A function that takes a schema and returns a middleware function. Reusable across routes.
3. How do you compose validators in Express?
Pass multiple middleware functions to the route: router.post('/', validateBody, sanitize, handler).
4. What is the error handling middleware?
A middleware with 4 parameters that catches errors from previous middleware and formats the response.
Challenge
Build an Express-style validation system with: schema-based validator factory, body/query/params validators, sanitization middleware, global error handler, and route composition.
FAQ
Mini Project: Express Validation Builder
# express_builder.py
from typing import Any, Callable, Dict
class ValidationBuilder:
def __init__(self):
self._rules: Dict[str, Dict] = {}
def field(self, name: str, typ: type = None, required: bool = False):
self._rules[name] = {"type": typ, "required": required}
return self
def build(self) -> Callable:
rules = dict(self._rules)
def middleware(req, res, next_fn):
errors = []
for name, rule in rules.items():
val = req.body.get(name)
if rule["required"] and val is None:
errors.append({"field": name, "code": "required"})
elif val is not None and rule["type"] and not isinstance(val, rule["type"]):
errors.append({"field": name, "code": "type_error"})
if errors:
res.status(400).json({"errors": errors})
else:
next_fn()
return middleware
builder = ValidationBuilder().field("email", str, True).field("age", int, True)
validate = builder.build()
req = Request()
req.body = {"email": "a@x.com", "age": "thirty"}
res = Response()
validate(req, res, lambda: None)
print(res.body)
Expected output:
{'errors': [{'field': 'age', 'code': 'type_error'}]}
What's Next
You understand Express validation. Next, learn Joi validation, then build the validation mini project.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro