Coverage for src/lexigram/admin/forms/async_validation.py: 35%

1"""Async validation support for forms. 

2 

3This module provides async validation capabilities for 

4form fields and form-level validation. 

5 

6FORM-05: Async validation support. 

7""" 

8 

9from __future__ import annotations 

10 

11from dataclasses import dataclass, field 

12from typing import TYPE_CHECKING, Any, Protocol, TypeVar, runtime_checkable 

13 

14from lexigram.concurrency import Parallel 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Callable, Coroutine 

18 

19T = TypeVar("T") 

20 

21 

22try: 

23 from lexigram.contracts.exceptions import ValidationError as CoreValidationError 

24except ImportError: 

25 CoreValidationError = Exception # type: ignore[assignment,misc] 

26 

27 

28# ============================================================================ 

29# Protocols 

30# ============================================================================ 

31 

32 

33@runtime_checkable 

34class AsyncValidatorProtocol(Protocol): 

35 """Protocol for async validators.""" 

36 

37 async def validate( 

38 self, 

39 value: Any, 

40 context: dict[str, Any] | None = None, 

41 ) -> list[str]: ... 

42 

43 

44# ============================================================================ 

45# Validation Result 

46# ============================================================================ 

47 

48 

49@dataclass 

50class ValidationResult: 

51 """Result of validation operation.""" 

52 

53 is_valid: bool 

54 errors: dict[str, list[str]] = field(default_factory=dict) 

55 warnings: dict[str, list[str]] = field(default_factory=dict) 

56 

57 @classmethod 

58 def valid(cls) -> ValidationResult: 

59 """Create a valid result.""" 

60 return cls(is_valid=True) 

61 

62 @classmethod 

63 def invalid(cls, errors: dict[str, list[str]]) -> ValidationResult: 

64 """Create an invalid result.""" 

65 return cls(is_valid=False, errors=errors) 

66 

67 def add_error(self, field: str, message: str) -> None: 

68 """Add an error message.""" 

69 self.is_valid = False 

70 if field not in self.errors: 

71 self.errors[field] = [] 

72 self.errors[field].append(message) 

73 

74 def add_warning(self, field: str, message: str) -> None: 

75 """Add a warning message.""" 

76 if field not in self.warnings: 

77 self.warnings[field] = [] 

78 self.warnings[field].append(message) 

79 

80 def merge(self, other: ValidationResult) -> ValidationResult: 

81 """Merge with another validation result.""" 

82 merged_errors = {**self.errors} 

83 for field_name, messages in other.errors.items(): 

84 if field_name in merged_errors: 

85 merged_errors[field_name].extend(messages) 

86 else: 

87 merged_errors[field_name] = list(messages) 

88 

89 merged_warnings = {**self.warnings} 

90 for field_name, messages in other.warnings.items(): 

91 if field_name in merged_warnings: 

92 merged_warnings[field_name].extend(messages) 

93 else: 

94 merged_warnings[field_name] = list(messages) 

95 

96 return ValidationResult( 

97 is_valid=self.is_valid and other.is_valid, 

98 errors=merged_errors, 

99 warnings=merged_warnings, 

100 ) 

101 

102 

103# ============================================================================ 

104# Async Validators 

105# ============================================================================ 

106 

107 

108class AsyncValidator: 

109 """Base class for async validators. 

110 

111 Example: 

112 >>> class UniqueEmailValidator(AsyncValidator): 

113 ... async def validate(self, value, context=None): 

114 ... if await check_email_exists(value): 

115 ... return ["Email already exists"] 

116 ... return [] 

117 """ 

118 

119 async def validate( 

120 self, 

121 value: Any, 

122 context: dict[str, Any] | None = None, 

123 ) -> list[str]: 

124 """Validate value asynchronously. 

125 

126 Args: 

127 value: Value to validate 

128 context: Optional validation context 

129 

130 Returns: 

131 List of error messages (empty if valid) 

132 """ 

133 raise NotImplementedError 

134 

135 

136class UniqueValidator(AsyncValidator): 

137 """Validator to check uniqueness. 

138 

139 Example: 

140 >>> validator = UniqueValidator( 

141 ... check_func=lambda v: User.exists(email=v), 

142 ... message="Email already registered" 

143 ... ) 

144 """ 

145 

146 def __init__( 

147 self, 

148 check_func: Callable[[Any], Coroutine[Any, Any, bool]], 

149 message: str = "Value already exists", 

150 exclude_id: str | None = None, 

151 ): 

152 self._check_func = check_func 

153 self._message = message 

154 self._exclude_id = exclude_id 

155 

156 async def validate( 

157 self, 

158 value: Any, 

159 context: dict[str, Any] | None = None, 

160 ) -> list[str]: 

161 context = context or {} 

162 

163 # Skip if checking for update and value matches original 

164 if self._exclude_id and context.get("id"): 

165 exclude_value = context.get(self._exclude_id) 

166 if exclude_value == value: 

167 return [] 

168 

169 exists = await self._check_func(value) 

170 if exists: 

171 return [self._message] 

172 return [] 

173 

174 

175class RemoteValidator(AsyncValidator): 

176 """Validator that calls a remote service. 

177 

178 Example: 

179 >>> validator = RemoteValidator( 

180 ... url="https://api.example.com/validate", 

181 ... error_key="errors" 

182 ... ) 

183 """ 

184 

185 def __init__( 

186 self, 

187 validate_func: Callable[[Any], Coroutine[Any, Any, dict[str, Any]]], 

188 error_key: str = "errors", 

189 ): 

190 self._validate_func = validate_func 

191 self._error_key = error_key 

192 

193 async def validate( 

194 self, 

195 value: Any, 

196 context: dict[str, Any] | None = None, 

197 ) -> list[str]: 

198 result = await self._validate_func(value) 

199 errors = result.get(self._error_key, []) 

200 if isinstance(errors, str): 

201 errors = [errors] 

202 return errors 

203 

204 

205class CompositeAsyncValidator(AsyncValidator): 

206 """Combines multiple async validators. 

207 

208 Example: 

209 >>> validator = CompositeAsyncValidator([ 

210 ... UniqueEmailValidator(), 

211 ... DomainValidator(), 

212 ... ]) 

213 """ 

214 

215 def __init__( 

216 self, 

217 validators: list[AsyncValidator], 

218 stop_on_first_error: bool = False, 

219 ): 

220 self._validators = validators 

221 self._stop_on_first_error = stop_on_first_error 

222 

223 async def validate( 

224 self, 

225 value: Any, 

226 context: dict[str, Any] | None = None, 

227 ) -> list[str]: 

228 all_errors: list[str] = [] 

229 

230 if self._stop_on_first_error: 

231 for validator in self._validators: 

232 errors = await validator.validate(value, context) 

233 if errors: 

234 return errors 

235 else: 

236 # Run all validators in parallel 

237 tasks = [v.validate(value, context) for v in self._validators] 

238 results = await Parallel.gather(*tasks) 

239 for errors in results: 

240 all_errors.extend(errors) 

241 

242 return all_errors 

243 

244 

245# ============================================================================ 

246# Async Field Validator 

247# ============================================================================ 

248 

249 

250@dataclass 

251class AsyncFieldConfig: 

252 """Configuration for async field validation.""" 

253 

254 field_name: str 

255 validators: list[AsyncValidator] = field(default_factory=list) 

256 debounce_ms: int = 300 

257 validate_on_change: bool = True 

258 validate_on_blur: bool = True 

259 

260 

261class AsyncFieldValidator: 

262 """Async validator for individual form fields. 

263 

264 Example: 

265 >>> validator = AsyncFieldValidator( 

266 ... field_name="email", 

267 ... validators=[UniqueEmailValidator()], 

268 ... ) 

269 >>> errors = await validator.validate("[email protected]") 

270 """ 

271 

272 def __init__( 

273 self, 

274 field_name: str, 

275 validators: list[AsyncValidator] | None = None, 

276 ): 

277 self.field_name = field_name 

278 self.validators = validators or [] 

279 

280 def add_validator(self, validator: AsyncValidator) -> AsyncFieldValidator: 

281 """Add a validator.""" 

282 self.validators.append(validator) 

283 return self 

284 

285 async def validate( 

286 self, 

287 value: Any, 

288 context: dict[str, Any] | None = None, 

289 ) -> list[str]: 

290 """Validate field value.""" 

291 all_errors: list[str] = [] 

292 

293 for validator in self.validators: 

294 errors = await validator.validate(value, context) 

295 all_errors.extend(errors) 

296 

297 return all_errors 

298 

299 

300# ============================================================================ 

301# Async Form Validator 

302# ============================================================================ 

303 

304 

305class AsyncFormValidator: 

306 """Async validator for entire forms. 

307 

308 Validates all fields in parallel for efficiency. 

309 

310 Example: 

311 >>> validator = AsyncFormValidator() 

312 >>> validator.field("email").unique(check_email_exists) 

313 >>> validator.field("username").unique(check_username_exists) 

314 >>> result = await validator.validate(form_data) 

315 """ 

316 

317 def __init__(self) -> None: 

318 self._field_validators: dict[str, AsyncFieldValidator] = {} 

319 self._form_validators: list[ 

320 Callable[[dict], Coroutine[Any, Any, list[str]]] 

321 ] = [] 

322 

323 def field(self, field_name: str) -> AsyncFieldValidatorBuilder: 

324 """Get or create field validator builder.""" 

325 if field_name not in self._field_validators: 

326 self._field_validators[field_name] = AsyncFieldValidator(field_name) 

327 return AsyncFieldValidatorBuilder(self._field_validators[field_name]) 

328 

329 def add_form_validator( 

330 self, 

331 validator: Callable[[dict], Coroutine[Any, Any, list[str]]], 

332 ) -> AsyncFormValidator: 

333 """Add form-level validator.""" 

334 self._form_validators.append(validator) 

335 return self 

336 

337 async def validate( 

338 self, 

339 data: dict[str, Any], 

340 context: dict[str, Any] | None = None, 

341 ) -> ValidationResult: 

342 """Validate form data. 

343 

344 All field validations run in parallel. 

345 """ 

346 result = ValidationResult.valid() 

347 context = context or {} 

348 

349 # Validate fields in parallel 

350 field_tasks: list[tuple[str, Any]] = [] 

351 for field_name, validator in self._field_validators.items(): 

352 value = data.get(field_name) 

353 field_tasks.append((field_name, validator.validate(value, context))) 

354 

355 if field_tasks: 

356 field_results = await Parallel.gather( 

357 *[t[1] for t in field_tasks], 

358 ) 

359 for (field_name, _), errors in zip( 

360 field_tasks, 

361 field_results, 

362 strict=False, 

363 ): 

364 for error in errors: 

365 result.add_error(field_name, error) 

366 

367 # Form-level validators 

368 for validator in self._form_validators: # type: ignore[assignment] 

369 errors = await validator(data) # type: ignore[operator] 

370 for error in errors: 

371 result.add_error("__form__", error) 

372 

373 return result 

374 

375 async def validate_field( 

376 self, 

377 field_name: str, 

378 value: Any, 

379 context: dict[str, Any] | None = None, 

380 ) -> list[str]: 

381 """Validate single field (for live validation).""" 

382 if field_name in self._field_validators: 

383 return await self._field_validators[field_name].validate(value, context) 

384 return [] 

385 

386 

387class AsyncFieldValidatorBuilder: 

388 """Builder for async field validators.""" 

389 

390 def __init__(self, field_validator: AsyncFieldValidator): 

391 self._validator = field_validator 

392 

393 def unique( 

394 self, 

395 check_func: Callable[[Any], Coroutine[Any, Any, bool]], 

396 message: str = "Value already exists", 

397 ) -> AsyncFieldValidatorBuilder: 

398 """Add uniqueness validator.""" 

399 self._validator.add_validator(UniqueValidator(check_func, message)) 

400 return self 

401 

402 def remote( 

403 self, 

404 validate_func: Callable[[Any], Coroutine[Any, Any, dict[str, Any]]], 

405 error_key: str = "errors", 

406 ) -> AsyncFieldValidatorBuilder: 

407 """Add remote validator.""" 

408 self._validator.add_validator(RemoteValidator(validate_func, error_key)) 

409 return self 

410 

411 def custom(self, validator: AsyncValidator) -> AsyncFieldValidatorBuilder: 

412 """Add custom validator.""" 

413 self._validator.add_validator(validator) 

414 return self 

415 

416 

417# ============================================================================ 

418# Decorator for async validation 

419# ============================================================================ 

420 

421 

422def async_validate( 

423 validator: AsyncFormValidator, 

424 on_error: str = "raise", 

425) -> Callable[[Callable[..., Any]], Callable[..., Any]]: 

426 """Decorator to add async validation to form handlers. 

427 

428 Args: 

429 validator: Form validator instance 

430 on_error: "raise" to raise exception, "return" to return result 

431 

432 Example: 

433 >>> validator = AsyncFormValidator() 

434 >>> validator.field("email").unique(check_email) 

435 >>> 

436 >>> @async_validate(validator) 

437 ... async def create_user(data: dict) -> User: 

438 ... return await User.create(**data) 

439 """ 

440 from functools import wraps 

441 

442 def decorator(func: Callable[..., Any]) -> Callable[..., Any]: 

443 @wraps(func) 

444 async def wrapper(data: dict[str, Any], *args: Any, **kwargs: Any) -> Any: 

445 result = await validator.validate(data) 

446 

447 if not result.is_valid: 

448 if on_error == "raise": 

449 raise AsyncValidationError(result) 

450 return result 

451 

452 return await func(data, *args, **kwargs) 

453 

454 return wrapper 

455 

456 return decorator 

457 

458 

459class AsyncValidationError(CoreValidationError): 

460 """Raised when async validation fails.""" 

461 

462 _code: str = "LEX_ERR_ADMIN_029" 

463 

464 def __init__(self, result: ValidationResult, **kwargs: Any) -> None: 

465 self.result = result 

466 super().__init__( 

467 f"Validation failed: {result.errors}", 

468 details={"errors": result.errors}, 

469 **kwargs, 

470 ) 

471 

472 

473__all__ = [ 

474 # Field validation 

475 "AsyncFieldConfig", 

476 "AsyncFieldValidator", 

477 "AsyncFieldValidatorBuilder", 

478 # Form validation 

479 "AsyncFormValidator", 

480 # Exceptions 

481 "AsyncValidationError", 

482 # Validators 

483 "AsyncValidator", 

484 "AsyncValidatorProtocol", 

485 "CompositeAsyncValidator", 

486 "RemoteValidator", 

487 "UniqueValidator", 

488 # Results 

489 "ValidationResult", 

490 # Decorator 

491 "async_validate", 

492]