Middleware Validation — Complete Guide
In this tutorial, you'll learn about Middleware Validation. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Middleware validation intercepts requests before route handlers, running validation logic and either passing the request through or returning validation errors.
What You'll Learn
By the end of this lesson, you will implement Express-style validation middleware, compose multiple middleware validators, and attach validation results to the request context.
Why It Matters
Middleware is the standard pattern for validation in web frameworks. It keeps validation separate from business logic and reusable across routes.
Real-World Use
Express.js applications use middleware stacks: app.post('/users', validateUser, createUser). The validateUser middleware checks the body before createUser runs.
Middleware Validation Flow
sequenceDiagram
Client->>Middleware: Request
Middleware->>Middleware: Validate Data
Middleware->>Middleware: Valid?
Middleware->>Handler: Pass to next()
Middleware->>Client: Or return 400
Basic Validation Middleware
# validation_middleware.py
from typing import Any, Callable, Dict, List, Optional
class Request:
def __init__(self, body: Dict = None, params: Dict = None, query: Dict = None):
self.body = body or {}
self.params = params or {}
self.query = query or {}
self.validation_errors: List[str] = []
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: Dict):
self.body = data
return self
class NextFunction:
def __init__(self):
self.called = False
def __call__(self):
self.called = True
def validate_body(schema: Dict[str, type]):
def middleware(req: Request, res: Response, next_fn: NextFunction):
errors = []
for field, field_type in schema.items():
value = req.body.get(field)
if value is None:
errors.append(f"{field}: required")
elif not isinstance(value, field_type):
errors.append(f"{field}: expected {field_type.__name__}")
if errors:
res.status(400).json({"error": "Validation failed", "details": errors})
return
next_fn()
return middleware
def handler(req: Request, res: Response, next_fn: NextFunction):
res.status(200).json({"message": "User created", "user": req.body})
req = Request(body={"name": "Alice", "age": 30})
invalid_req = Request(body={"name": "Alice"})
user_schema = {"name": str, "age": int, "email": str}
res = Response()
next_fn = NextFunction()
validate_body(user_schema)(req, res, next_fn)
print(f"Valid: next called={next_fn.called}, status={res.status_code}")
res2 = Response()
next_fn2 = NextFunction()
validate_body(user_schema)(invalid_req, res2, next_fn2)
print(f"Invalid: next called={next_fn2.called}, status={res2.status_code}, body={res2.body}")
Expected output:
Valid: next called=True, status=200
Invalid: next called=False, status=400, body={'error': 'Validation failed', 'details': ['email: required']}
Composable Middleware
# composable_middleware.py
from typing import Any, Callable, Dict, List
class MiddlewareChain:
def __init__(self):
self.middleware: List[Callable] = []
def use(self, middleware_fn: Callable):
self.middleware.append(middleware_fn)
def run(self, req: Dict) -> Dict:
ctx = {"req": req, "res": {}, "errors": [], "next_called": False}
def next_fn():
ctx["next_called"] = True
for mw in self.middleware:
if ctx["errors"]:
break
mw(ctx, next_fn)
if not ctx["next_called"]:
break
ctx["next_called"] = False
return ctx
def parse_json(ctx, next_fn):
try:
ctx["parsed"] = ctx["req"]
next_fn()
except Exception as e:
ctx["errors"].append(str(e))
def validate_name(ctx, next_fn):
name = ctx.get("parsed", {}).get("name", "")
if len(name) < 2:
ctx["errors"].append("name too short")
else:
next_fn()
def validate_age(ctx, next_fn):
age = ctx.get("parsed", {}).get("age", 0)
if not isinstance(age, (int, float)) or age < 0:
ctx["errors"].append("invalid age")
else:
next_fn()
chain = MiddlewareChain()
chain.use(parse_json)
chain.use(validate_name)
chain.use(validate_age)
result = chain.run({"name": "Alice", "age": 30})
print(f"Valid: errors={result['errors']}")
result2 = chain.run({"name": "A", "age": -1})
print(f"Invalid: errors={result2['errors']}")
Expected output:
Valid: errors=[]
Invalid: errors=['name too short', 'invalid age']
Route-Specific Validation
# route_validation.py
from typing import Any, Callable, Dict
class Router:
def __init__(self):
self.routes: Dict[str, Dict] = {}
def post(self, path: str, *middleware: Callable):
self.routes[("POST", path)] = list(middleware)
def handle(self, method: str, path: str, body: Dict) -> Dict:
handlers = self.routes.get((method, path), [])
ctx = {"body": body, "errors": [], "next_called": False}
def next_fn():
ctx["next_called"] = True
for handler in handlers:
if ctx["errors"]:
return {"status": 400, "errors": ctx["errors"]}
ctx["next_called"] = False
handler(ctx, next_fn)
if not ctx["next_called"]:
break
if ctx["errors"]:
return {"status": 400, "errors": ctx["errors"]}
return {"status": 200, "data": body}
def validate_email(ctx, next_fn):
email = ctx["body"].get("email", "")
if "@" not in email:
ctx["errors"].append("invalid email")
else:
next_fn()
def validate_password(ctx, next_fn):
pwd = ctx["body"].get("password", "")
if len(pwd) < 8:
ctx["errors"].append("password too short")
else:
next_fn()
router = Router()
router.post("/register", validate_email, validate_password)
print(router.handle("POST", "/register", {"email": "a@x.com", "password": "secret123"}))
print(router.handle("POST", "/register", {"email": "invalid", "password": "short"}))
Expected output:
{'status': 200, 'data': {'email': 'a@x.com', 'password': 'secret123'}}
{'status': 400, 'errors': ['invalid email', 'password too short']}
Common Mistakes
1. Running Validation After Business Logic
Validation must happen before the handler. Handler should assume data is already valid.
2. Not Calling next() Correctly
Forgetting to call next() after successful validation blocks the request permanently.
3. Mutating Request in Middleware
Sanitization middleware should clone or create a new object, not mutate the original request.
4. Too Much Logic in Middleware
Middleware should validate, not transform data. Heavy transformations belong in the handler or service layer.
5. No Error Return Path
Middleware must return a proper error response when validation fails, not just log a warning.
Practice Questions
1. What is middleware validation?
Validation logic that runs between the request and the route handler, intercepting invalid data.
2. What happens if validation middleware fails?
It returns a 400 response with error details and does not call next().
3. How do you compose multiple validators?
Chain middleware functions. Each validator calls next() on success or returns errors on failure.
4. Why is middleware validation reusable?
A validateBody(schema) function can be used on any route by passing it in the middleware chain.
Challenge
Build a middleware validation system that supports validating body, query params, and URL params simultaneously, with proper error aggregation.
FAQ
Mini Project: Middleware Validator
# middleware_validator.py
from typing import Any, Callable, Dict, List
class MiddlewareValidator:
def __init__(self):
self.validators: List[Callable] = []
def add(self, validator: Callable):
self.validators.append(validator)
def run(self, data: Dict) -> List[str]:
errors = []
for validator in self.validators:
err = validator(data)
if err:
errors.append(err)
return errors
validator = MiddlewareValidator()
validator.add(lambda d: "name required" if not d.get("name") else None)
validator.add(lambda d: "bad email" if "@" not in d.get("email", "") else None)
validator.add(lambda d: "age must be > 0" if d.get("age", 0) <= 0 else None)
print(validator.run({"name": "Alice", "email": "a@x.com", "age": 30}))
print(validator.run({"name": "", "email": "bad", "age": -1}))
Expected output:
[]
['name required', 'bad email', 'age must be > 0']
What's Next
You understand middleware validation. Next, learn input sanitization, then type coercion.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro