Coverage for src/lexigram/admin/forms/validation.py: 78%
96 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Validation engine for Lexigram Admin Forms."""
3from __future__ import annotations
5import asyncio
6from dataclasses import dataclass
7import re
8from typing import TYPE_CHECKING, Any, Protocol
10from lexigram.concurrency import Parallel
12if TYPE_CHECKING:
13 from collections.abc import Callable
16@dataclass
17class ValidationError:
18 """Represents a validation failure.
20 Attributes:
21 message: Human-readable error message.
22 code: Optional error code for machine consumption.
23 """
25 message: str
26 code: str | None = None
29ValidatorResult = str | list[str] | ValidationError | list[ValidationError] | None
32class Validator(Protocol):
33 """Protocol for form validators.
35 Can be a simple function or a class with __call__.
36 Supports both sync and async execution.
37 """
39 async def __call__(
40 self,
41 value: Any,
42 context: dict[str, Any],
43 ) -> ValidatorResult: ...
46class FormValidationEngine:
47 """Engine for executing complex form validation logic.
49 Supports field-level validators and form-level (cross-field) validators.
50 """
52 def __init__(self) -> None:
53 """Initialize empty validator maps."""
54 self._field_validators: dict[str, list[Validator]] = {}
55 self._form_validators: list[Validator] = []
57 def add_field_validator(self, field_name: str, validator: Validator) -> None:
58 """Add a validator to a specific field."""
59 if field_name not in self._field_validators:
60 self._field_validators[field_name] = []
61 self._field_validators[field_name].append(validator)
63 def add_form_validator(self, validator: Validator) -> None:
64 """Add a validator for the entire form (cross-field)."""
65 self._form_validators.append(validator)
67 async def validate_field(
68 self,
69 field_name: str,
70 value: Any,
71 data: dict[str, Any],
72 ) -> list[ValidationError]:
73 """Validate a single field's value."""
74 validators = self._field_validators.get(field_name, [])
75 errors = []
77 for validator in validators:
78 result = await self._run_validator(validator, value, data)
79 if result:
80 errors.extend(self._normalize_errors(result))
82 return errors
84 async def validate_form(
85 self,
86 data: dict[str, Any],
87 ) -> dict[str, list[ValidationError]]:
88 """Validate an entire form data set.
90 Runs all field validators and then form-level validators.
91 """
92 all_errors: dict[str, list[ValidationError]] = {}
94 # 1. Field-level validation
95 field_tasks = []
96 field_names = list(set(list(self._field_validators.keys()) + list(data.keys())))
98 for name in field_names:
99 val = data.get(name)
100 field_tasks.append(self.validate_field(name, val, data))
102 field_results = await Parallel.gather(*field_tasks)
103 all_errors = {
104 name: errors
105 for name, errors in zip(field_names, field_results, strict=False)
106 if errors
107 }
109 # 2. Form-level (cross-field) validation
110 form_tasks = [self._run_validator(v, None, data) for v in self._form_validators]
111 form_results = await Parallel.gather(*form_tasks)
113 for result in form_results:
114 if result:
115 # Form level errors usually apply to the whole form or multiple fields.
116 # If the result is a dict, we map to fields, otherwise to "__all__".
117 if isinstance(result, dict):
118 for field_name, field_errors in result.items():
119 if field_name not in all_errors:
120 all_errors[field_name] = []
121 all_errors[field_name].extend(
122 self._normalize_errors(field_errors),
123 )
124 else:
125 if "__all__" not in all_errors:
126 all_errors["__all__"] = []
127 all_errors["__all__"].extend(self._normalize_errors(result))
129 return all_errors
131 async def _run_validator(
132 self,
133 validator: Any,
134 value: Any,
135 data: dict[str, Any],
136 ) -> Any:
137 """Execute a validator regardless of whether it's sync or async."""
138 if asyncio.iscoroutinefunction(validator):
139 return await validator(value, data)
140 if callable(validator):
141 # Check if it's a class instance with an async __call__
142 if callable(validator) and asyncio.iscoroutinefunction(
143 validator.__call__,
144 ):
145 return await validator(value, data)
146 return validator(value, data)
147 return None
149 def _normalize_errors(self, result: ValidatorResult) -> list[ValidationError]:
150 """Convert various validator return types into a standard list of errors."""
151 if result is None:
152 return []
153 if isinstance(result, str):
154 return [ValidationError(result)]
155 if isinstance(result, ValidationError):
156 return [result]
157 if isinstance(result, list):
158 normalized = []
159 for item in result:
160 normalized.extend(self._normalize_errors(item))
161 return normalized
162 return []
165# Standard Validators
168def required(value: Any, _: dict[str, Any]) -> ValidatorResult:
169 """Verify that a value is not empty."""
170 if value is None or (isinstance(value, str) and not value.strip()):
171 return "This field is required."
172 return None
175def email(value: Any, _: dict[str, Any]) -> ValidatorResult:
176 """Verify that a value is a valid email address."""
177 if not value:
178 return None
179 pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
180 if not re.match(pattern, str(value)):
181 return "Please enter a valid email address."
182 return None
185def min_length(length: int) -> Callable:
186 """Factory for minimum length validation."""
188 def validator(value: Any, _: dict[str, Any]) -> ValidatorResult:
189 if value and len(str(value)) < length:
190 return f"Must be at least {length} characters long."
191 return None
193 return validator
196def max_length(length: int) -> Callable:
197 """Factory for maximum length validation."""
199 def validator(value: Any, _: dict[str, Any]) -> ValidatorResult:
200 if value and len(str(value)) > length:
201 return f"Cannot exceed {length} characters."
202 return None
204 return validator