Data Validation Project — Build a Complete Validation System
DodaTech
Updated 2026-06-28
11 min read
In this tutorial, you'll build a complete Data Validation System. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
This project combines all validation concepts into a production-ready validation system with middleware, custom rules, cross-field checks, security sanitization, and error formatting.
What You'll Learn
By the end of this lesson, you will build a complete validation framework with rule definitions, middleware integration, error collection and formatting, and security-aware input processing.
Why It Matters
Individual validation lessons teach components. This project connects them into a working system that handles real-world validation for user registration, product management, and API requests.
Real-World Use
Durga Antivirus Pro uses a similar validation pipeline for its license activation API: type checking, format validation, custom license key checksum, cross-field tier validation, and Rate Limiting — all in one pipeline.
System Architecture
flowchart TD
Request[HTTP Request] --> Router[Router]
Router --> Middleware[Validation Middleware]
Middleware --> Schema[Schema Definition]
Schema --> Rules{Rules Engine}
Rules --> Type[Type Check]
Rules --> Format[Format Check]
Rules --> Range[Range Check]
Rules --> Custom[Custom Rules]
Rules --> CrossField[Cross-Field Rules]
Rules --> Security[Security Checks]
Rules --> Error{Errors Found?}
Error -->|Yes| Formatter[Error Formatter]
Formatter --> Response[Error Response]
Error -->|No| Sanitizer[Sanitization]
Sanitizer --> Controller[Controller]
Controller --> Success[Success Response]
Core Validation Framework
# validation_framework.py
from typing import Any, Callable, Dict, List, Optional, Tuple
class ValidationError:
def __init__(self, field: str, code: str, message: str):
self.field = field
self.code = code
self.message = message
def to_dict(self) -> Dict:
return {"field": self.field, "code": self.code, "message": self.message}
class ValidationRule:
def __init__(self, name: str, fields: List[str],
validate: Callable[[Dict], Optional[ValidationError]],
depends_on: Optional[List[str]] = None):
self.name = name
self.fields = fields
self.validate = validate
self.depends_on = depends_on or []
class ValidationSchema:
def __init__(self, name: str):
self.name = name
self.rules: List[ValidationRule] = []
def add_rule(self, rule: ValidationRule):
self.rules.append(rule)
def validate(self, data: Dict) -> List[ValidationError]:
errors = []
for rule in self.rules:
error = rule.validate(data)
if error:
errors.append(error)
return errors
class ValidationResult:
def __init__(self):
self.errors: List[ValidationError] = []
def add_error(self, error: ValidationError):
self.errors.append(error)
def is_valid(self) -> bool:
return len(self.errors) == 0
def to_response(self) -> Dict:
return {
"valid": self.is_valid(),
"count": len(self.errors),
"errors": [e.to_dict() for e in self.errors]
}
schema = ValidationSchema("user_registration")
schema.add_rule(ValidationRule("required_fields", ["username", "email", "password"],
lambda d: next((ValidationError(f, "required", f"{f} is required")
for f in ["username", "email", "password"] if not d.get(f)), None)))
schema.add_rule(ValidationRule("email_format", ["email"],
lambda d: ValidationError("email", "format", "Invalid email format")
if d.get("email") and "@" not in str(d["email"]) else None))
result = schema.validate({"username": "alice", "email": "bad-email", "password": ""})
print(f"Schema: {schema.name}")
print(f"Valid: {result.is_valid()}")
for err in result.errors:
print(f" {err.field}: [{err.code}] {err.message}")
Expected output:
Schema: user_registration
Valid: False
email: [format] Invalid email format
password: [required] password is required
Validation Middleware
# validation_middleware.py
from typing import Any, Callable, Dict, List, Optional
class Request:
def __init__(self, body: Dict, headers: Dict = None, params: Dict = None):
self.body = body
self.headers = headers or {}
self.params = params or {}
class Response:
def __init__(self):
self.status_code = 200
self.body: Dict = {}
def json(self, data: Dict, status: int = 200):
self.body = data
self.status_code = status
return self
class ValidationMiddleware:
def __init__(self, schema: Any):
self.schema = schema
def before(self, request: Request) -> Optional[Response]:
errors = self.schema.validate(request.body)
if not errors.is_valid():
return Response().json(errors.to_response(), status=422)
return None
class RouteHandler:
def __init__(self):
self.middlewares: List[ValidationMiddleware] = []
def use(self, middleware: ValidationMiddleware):
self.middlewares.append(middleware)
def handle(self, request: Request, handler: Callable[[Request], Response]) -> Response:
for mw in self.middlewares:
error_response = mw.before(request)
if error_response:
return error_response
return handler(request)
schema = ValidationSchema("create_user")
schema.add_rule(ValidationRule("name_required", ["name"],
lambda d: ValidationError("name", "required", "Name is required")
if not d.get("name") else None))
mw = ValidationMiddleware(schema)
router = RouteHandler()
router.use(mw)
def create_user_handler(req: Request) -> Response:
return Response().json({"message": f"User {req.body.get('name')} created"}, 201)
valid_req = Request({"name": "Alice"})
resp = router.handle(valid_req, create_user_handler)
print(f"Valid request: status={resp.status_code} body={resp.body}")
invalid_req = Request({})
resp2 = router.handle(invalid_req, create_user_handler)
print(f"Invalid request: status={resp2.status_code} body={resp2.body}")
Expected output:
Valid request: status=201 body={'message': 'User Alice created'}
Invalid request: status=422 body={'valid': False, 'count': 1, 'errors': [{'field': 'name', 'code': 'required', 'message': 'Name is required'}]}
Security-Aware Validation
# secure_validation.py
import re
from typing import Any, Dict, List, Optional
class SecurityValidator:
DANGEROUS_SQL = re.compile(r"['\";\-\-]|DROP\s+TABLE|UNION\s+SELECT|OR\s+1\s*=\s*1", re.I)
DANGEROUS_HTML = re.compile(r"<script[^>]*>|<[^>]*onerror|<[^>]*onload|javascript:", re.I)
DANGEROUS_PATH = re.compile(r"\.\./|\.\.\\|[\\/]etc[\\/]")
@staticmethod
def sanitize_string(value: str) -> str:
value = value.strip()
value = value.replace("<", "<").replace(">", ">")
value = value.replace("\"", """).replace("'", "'")
return value
@staticmethod
def validate_no_injection(value: str, field: str) -> Optional[ValidationError]:
if SecurityValidator.DANGEROUS_SQL.search(value):
return ValidationError(field, "sql_injection", "Input contains SQL metacharacters")
if SecurityValidator.DANGEROUS_HTML.search(value):
return ValidationError(field, "xss", "Input contains disallowed HTML")
if SecurityValidator.DANGEROUS_PATH.search(value):
return ValidationError(field, "path_traversal", "Input contains path traversal")
return None
@staticmethod
def secure_username(value: str) -> Optional[ValidationError]:
if not re.match(r'^[a-zA-Z0-9_]{3,30}$', value):
return ValidationError("username", "format", "Username: 3-30 alphanumeric chars or underscores")
return SecurityValidator.validate_no_injection(value, "username")
@staticmethod
def secure_comment(value: str, max_len: int = 1000) -> Optional[ValidationError]:
if len(value) > max_len:
return ValidationError("comment", "max_length", f"Comment max {max_len} chars")
inj_error = SecurityValidator.validate_no_injection(value, "comment")
if inj_error:
return inj_error
return None
sv = SecurityValidator()
tests = [
("alice_dev", sv.secure_username),
("admin' --", sv.secure_username),
("<script>alert(1)</script>", sv.secure_comment),
("../../../etc/passwd", sv.secure_comment),
]
for value, validator in tests:
error = validator(value)
safe = sv.sanitize_string(value)
status = "VALID" if not error else f"[{error.code}] {error.message}"
print(f" Input: {str(value):35s} Status: {status}")
print(f" Sanitized: {safe}")
print()
Expected output:
Input: alice_dev Status: VALID
Sanitized: alice_dev
Input: admin' -- Status: [sql_injection] Input contains SQL metacharacters
Sanitized: admin' --
Input: <script>alert(1)</script> Status: [xss] Input contains disallowed HTML
Sanitized: <script>alert(1)</script>
Input: ../../../etc/passwd Status: [path_traversal] Input contains path traversal
Sanitized: ../../../etc/passwd
Combined Validation System
# combined_system.py
import re
from typing import Any, Callable, Dict, List, Optional, Tuple
from datetime import datetime
class ValidationEngine:
def __init__(self):
self.schemas: Dict[str, ValidationSchema] = {}
self.global_rules: List[ValidationRule] = []
def register_schema(self, schema: ValidationSchema):
self.schemas[schema.name] = schema
def add_global_rule(self, rule: ValidationRule):
self.global_rules.append(rule)
def validate(self, schema_name: str, data: Dict) -> ValidationResult:
result = ValidationResult()
for rule in self.global_rules:
error = rule.validate(data)
if error:
result.add_error(error)
schema = self.schemas.get(schema_name)
if schema:
for rule in schema.rules:
error = rule.validate(data)
if error:
result.add_error(error)
return result
engine = ValidationEngine()
# Global security rule
engine.add_global_rule(ValidationRule("xss_check", ["*"],
lambda d: next((ValidationError(k, "xss", "HTML tags not allowed in " + k)
for k, v in d.items() if isinstance(v, str) and re.search(r'<[^>]*>', v)), None)))
# Registration schema
reg_schema = ValidationSchema("register")
reg_schema.add_rule(ValidationRule("required", ["username", "email", "password"],
lambda d: next((ValidationError(f, "required", f"{f} required")
for f in ["username", "email", "password"] if not d.get(f)), None)))
reg_schema.add_rule(ValidationRule("email_format", ["email"],
lambda d: ValidationError("email", "format", "Invalid email")
if d.get("email") and "@" not in str(d["email"]) else None))
reg_schema.add_rule(ValidationRule("password_strength", ["password"],
lambda d: ValidationError("password", "weak", "Min 8 chars, 1 upper, 1 digit, 1 special")
if d.get("password") and (len(d["password"]) < 8 or
not re.search(r'[A-Z]', d["password"]) or
not re.search(r'[0-9]', d["password"])) else None))
reg_schema.add_rule(ValidationRule("password_match", ["password", "confirm_password"],
lambda d: ValidationError("confirm_password", "mismatch", "Passwords do not match")
if d.get("password") and d.get("confirm_password") and
d["password"] != d["confirm_password"] else None))
engine.register_schema(reg_schema)
valid_data = {"username": "alice", "email": "a@x.com",
"password": "Str0ng!", "confirm_password": "Str0ng!"}
invalid_data = {"username": "<b>hacker</b>", "email": "bad",
"password": "weak", "confirm_password": "mismatch"}
for name, data in [("valid", valid_data), ("invalid", invalid_data)]:
result = engine.validate("register", data)
print(f"{name}: valid={result.is_valid()} errors={result.to_response()}")
print()
Expected output:
valid: valid=True errors={'valid': True, 'count': 0, 'errors': []}
invalid: valid=False errors={'valid': False, 'count': 4, 'errors': [{'field': 'username', 'code': 'xss', 'message': 'HTML tags not allowed in username'}, {'field': 'email', 'code': 'format', 'message': 'Invalid email'}, {'field': 'password', 'code': 'weak', 'message': 'Min 8 chars, 1 upper, 1 digit, 1 special'}, {'field': 'confirm_password', 'code': 'mismatch', 'message': 'Passwords do not match'}]}
API Integration Example
# api_integration.py
from typing import Any, Callable, Dict, List, Optional
class APIApplication:
def __init__(self):
self.routes: Dict[str, Dict] = {}
def post(self, path: str, handler: Callable,
schema: Optional[ValidationSchema] = None):
self.routes[("POST", path)] = {"handler": handler, "schema": schema}
def handle_request(self, method: str, path: str, body: Dict) -> Dict:
route = self.routes.get((method, path))
if not route:
return {"status": 404, "body": {"error": "Not found"}}
if route["schema"]:
errors = route["schema"].validate(body)
if not errors.is_valid():
return {"status": 422, "body": errors.to_response()}
return route["handler"](body)
app = APIApplication()
login_schema = ValidationSchema("login")
login_schema.add_rule(ValidationRule("required_fields", ["email", "password"],
lambda d: next((ValidationError(f, "required", f"{f} required")
for f in ["email", "password"] if not d.get(f)), None)))
def login_handler(body: Dict) -> Dict:
return {"status": 200, "body": {"message": "Login successful", "user": body.get("email")}}
app.post("/login", login_handler, login_schema)
results = []
results.append(app.handle_request("POST", "/login", {"email": "user@x.com", "password": "pass123"}))
results.append(app.handle_request("POST", "/login", {"email": "user@x.com"}))
results.append(app.handle_request("POST", "/unknown", {}))
for r in results:
print(f"Status: {r['status']} Body: {r['body']}")
Expected output:
Status: 200 Body: {'message': 'Login successful', 'user': 'user@x.com'}
Status: 422 Body: {'valid': False, 'count': 1, 'errors': [{'field': 'password', 'code': 'required', 'message': 'password required'}]}
Status: 404 Body: {'error': 'Not found'}
Error Formatter with i18n
# error_formatter.py
from typing import Any, Dict, List, Optional
class I18nErrorFormatter:
def __init__(self):
self.messages = {
"en": {
"required": "{field} is required",
"format": "Invalid {field} format",
"weak": "Password is too weak",
"mismatch": "Passwords do not match",
"xss": "HTML tags are not allowed",
"min_length": "{field} must be at least {min} characters",
"max_length": "{field} must be at most {max} characters",
},
"es": {
"required": "{field} es obligatorio",
"format": "Formato de {field} invalido",
"weak": "La contrasena es muy debil",
"mismatch": "Las contrasenas no coinciden",
"xss": "Las etiquetas HTML no estan permitidas",
}
}
def format(self, errors: List[ValidationError], locale: str = "en") -> Dict:
locale_msgs = self.messages.get(locale, self.messages["en"])
formatted = []
for err in errors:
msg = locale_msgs.get(err.code, err.message)
try:
msg = msg.format(field=err.field)
except KeyError:
pass
formatted.append({"field": err.field, "code": err.code, "message": msg})
return {"valid": False, "count": len(formatted), "errors": formatted}
formatter = I18nErrorFormatter()
errors = [
ValidationError("email", "required", ""),
ValidationError("password", "weak", ""),
]
print("EN:", formatter.format(errors, "en"))
print("ES:", formatter.format(errors, "es"))
Expected output:
EN: {'valid': False, 'count': 2, 'errors': [{'field': 'email', 'code': 'required', 'message': 'email is required'}, {'field': 'password', 'code': 'weak', 'message': 'Password is too weak'}]}
ES: {'valid': False, 'count': 2, 'errors': [{'field': 'email', 'code': 'required', 'message': 'email es obligatorio'}, {'field': 'password', 'code': 'weak', 'message': 'La contrasena es muy debil'}]}
Common Mistakes
1. No Separation Of Concerns
Putting validation logic inside route handlers mixes responsibilities. Extract validation into middleware and schema definitions.
2. Global Rules That Are Too Broad
A global XSS check on all fields rejects valid input with angle brackets. Apply security checks selectively based on field type.
3. Stopping at the First Error
Returning the first error forces clients to fix and retry repeatedly. Collect all errors before responding.
4. Not Sanitizing After Validation
Validation passes or rejects input. Sanitization cleans valid input for safe use. Skipping sanitization leaves XSS and injection paths open.
5. Inconsistent Error Format Across Endpoints
Using different error formats (array vs object vs string) confuses clients. Standardize on one format across your entire API.
Practice Questions
1. What components does a validation system need?
Schema definitions, rule engine, middleware integration, error formatter, security validator, and optional localization.
2. Why use middleware for validation?
Middleware decouples validation from business logic. Routes stay clean, validation is testable, and all endpoints get consistent enforcement.
3. How do you handle cross-field rules in a pipeline?
Run individual field rules first, then cross-field rules that validate relationships between fields using the complete data object.
4. What is the difference between global and schema-specific rules?
Global rules apply to every request (security checks). Schema-specific rules apply only to specific endpoints.
Challenge
Build a complete REST API validation system with: schema registry, global security rules, per-endpoint schemas with custom rules, cross-field validation (password match, date range), error formatting with RFC 7807 and i18n support, middleware integration, and input sanitization. Support nested object validation with dot-notation field paths.
FAQ
Mini Project: Complete Validation System
# complete_validation_system.py
import re
from typing import Any, Callable, Dict, List, Optional, Tuple
class FieldValidator:
def __init__(self):
self.validators: Dict[str, Callable] = {}
def add(self, name: str, fn: Callable[[Any, str], Optional[str]]):
self.validators[name] = fn
def validate(self, field: str, value: Any, rules: List[str]) -> List[str]:
errors = []
for rule in rules:
fn = self.validators.get(rule)
if fn:
error = fn(value, field)
if error:
errors.append(error)
return errors
fv = FieldValidator()
fv.add("required", lambda v, f: f"{f} required" if v is None or v == "" else None)
fv.add("email", lambda v, f: f"Invalid {f}" if isinstance(v, str) and "@" not in v else None)
fv.add("min8", lambda v, f: f"{f} min 8 chars" if isinstance(v, str) and len(v) < 8 else None)
fv.add("no_html", lambda v, f: f"{f} no HTML" if isinstance(v, str) and "<" in v else None)
schema = {
"username": ["required", "no_html"],
"email": ["required", "email"],
"password": ["required", "min8"],
}
def validate_all(data: Dict, schema: Dict, field_validator: FieldValidator) -> Dict:
errors = {}
for field, rules in schema.items():
field_errors = field_validator.validate(field, data.get(field), rules)
if field_errors:
errors[field] = field_errors
return errors
data = {"username": "<script>xss</script>", "email": "bad", "password": "short"}
result = validate_all(data, schema, fv)
print(result)
Expected output:
{'username': ['username no HTML'], 'email': ['Invalid email'], 'password': ['password min 8 chars']}
What's Next
You have built a complete data validation system. Review all lessons: start here, or explore advanced type checking to deepen your understanding.
← Previous
Async Validation Patterns — Complete Guide
Next →
Conditional Validation Logic — Complete Guide
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro