Coverage for src/lexigram/web/routing/validation.py: 0%
274 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Request parameter validation helpers using Pydantic.
3Provides helpers to build a lightweight Pydantic model for query parameters and
4validate them in a single step. This reduces manual parsing in the router and
5provides clear validation errors.
6"""
8from __future__ import annotations
10from dataclasses import dataclass
11import inspect
12from typing import Any, get_args, get_origin
14from lexigram.contracts.exceptions.domain import ValidationError
15from lexigram.domain import DomainModel
16from lexigram.logging import get_logger
18logger = get_logger(__name__)
21def _create_validation_model(model_name: str, **fields: Any) -> type:
22 """Create a strict Pydantic validation model (not DomainModel).
24 Unlike create_model (which applies @dataclass and bypasses Pydantic's type
25 validation), this helper creates a plain pydantic.BaseModel so that type
26 coercion errors are raised as pydantic.ValidationError.
27 """
28 from pydantic import BaseModel, ConfigDict
29 from pydantic import create_model as _pm
31 class _ValidationBase(BaseModel):
32 model_config = ConfigDict(arbitrary_types_allowed=False)
34 return _pm(model_name, __base__=_ValidationBase, **fields)
37from functools import lru_cache
40# 256 entries covers virtually all real-world applications (typical apps have
41# well under 100 routes) while avoiding unbounded growth in pathological cases.
42@lru_cache(maxsize=256)
43def _cached_get_type_hints_for_handler(handler: Any) -> dict[str, Any]:
44 from typing import get_type_hints
46 # When a handler is wrapped by any decorator using @wraps,
47 # __annotations__ are copied but __globals__ points to the wrapper module.
48 # With `from __future__ import annotations`, string annotations need the
49 # original module's __globals__ to resolve. Follow __wrapped__ chain.
50 target = handler
51 while hasattr(target, "__wrapped__"):
52 target = target.__wrapped__
54 globalns = getattr(target, "__globals__", None) or getattr(
55 handler, "__globals__", None
56 )
57 try:
58 return get_type_hints(handler, globalns=globalns) or {}
59 except (NameError, AttributeError, TypeError):
60 # Fallback: try the unwrapped target directly
61 try:
62 return get_type_hints(target) or {}
63 except (NameError, AttributeError, TypeError):
64 return {}
67_SIMPLE_TYPES = {str, int, float, bool}
70def _get_cached_query_model(handler: Any) -> type[DomainModel] | None:
71 """Get cached query model from handler."""
72 cache_target = getattr(handler, "__func__", handler)
73 val = getattr(cache_target, "_query_model", None)
74 return val if isinstance(val, type) else None
77def _set_cached_query_model(handler: Any, model: type[DomainModel] | None) -> None:
78 """Set cached query model on handler."""
79 cache_target = getattr(handler, "__func__", handler)
80 cache_target._query_model = model
83def _get_cached_combined_model(handler: Any) -> type[DomainModel] | None:
84 """Get cached combined model from handler."""
85 cache_target = getattr(handler, "__func__", handler)
86 val = getattr(cache_target, "_combined_model", None)
87 return val if isinstance(val, type) else None
90def _set_cached_combined_model(handler: Any, model: type[DomainModel] | None) -> None:
91 """Set cached combined model on handler."""
92 cache_target = getattr(handler, "__func__", handler)
93 cache_target._combined_model = model
96def _is_simple_type(annotation: Any) -> bool:
97 """Return True if annotation represents a simple query-deserializable type."""
98 if annotation in _SIMPLE_TYPES:
99 return True
101 origin = get_origin(annotation)
102 if origin is type(None):
103 args = get_args(annotation)
104 if args and args[0] in _SIMPLE_TYPES:
105 return True
107 return False
110def get_or_build_query_model(handler: Any) -> type[DomainModel] | None:
111 """Build and cache a Pydantic model class for handler query parameters.
113 The model includes parameters that are simple types (str/int/float/bool)
114 and are not explicitly a Request, path param, or a Pydantic DomainModel
115 (which are treated as body models).
116 """
117 # When handler is a bound method, attributes can't be set on the method object.
118 # Use the underlying function object (handler.__func__) as the cache target when present.
119 cache_target = getattr(handler, "__func__", handler)
120 cached_model = _get_cached_query_model(handler)
121 if cached_model is not None:
122 return cached_model
124 sig = inspect.signature(handler)
125 hints = _cached_get_type_hints_for_handler(handler)
127 fields: dict[str, tuple] = {}
129 for name, param in sig.parameters.items():
130 if name == "self":
131 continue
133 annotation = hints.get(name, param.annotation)
135 # Skip request object or explicit DomainModel bodies
136 if annotation == inspect.Parameter.empty:
137 continue
139 # Skip Pydantic models (handled as body)
140 try:
141 if isinstance(annotation, type) and issubclass(annotation, DomainModel):
142 continue
143 except (TypeError, AttributeError) as e:
144 logger.debug("Annotation is not a type or cannot be inspected: %s", e)
146 if _is_simple_type(annotation):
147 default = (
148 param.default if param.default is not inspect.Parameter.empty else ...
149 )
150 fields[name] = (annotation, default)
152 if not fields:
153 _set_cached_query_model(handler, None)
154 return None
156 model_name = f"{getattr(cache_target, '__name__', 'Handler')}_QueryModel"
157 model = _create_validation_model(model_name, **fields)
158 _set_cached_query_model(handler, model)
159 return model
162def validate_query_params(request_query: dict[str, str], handler: Any) -> DomainModel:
163 """Validate and coerce query parameters using the handler's query model.
165 Raises ValidationError if validation fails.
166 """
167 model = get_or_build_query_model(handler)
169 if model is None:
170 # Nothing to validate
171 @dataclass(init=False)
172 class EmptyModel(DomainModel):
173 pass
175 return EmptyModel()
177 # Pydantic expects values of correct types; pass the dict of strings and let
178 # pydantic coerce types where possible
179 return model(**request_query)
182def get_or_build_combined_model(handler: Any) -> type[DomainModel] | None:
183 """Build and cache a Pydantic model class that merges path, query, and body params.
185 This creates a single model with fields corresponding to simple query/path
186 parameters and any parameters annotated as Pydantic models (these will be
187 nested models). This allows a single call to pydantic to validate and
188 coerce all inputs and produce a consistent ValidationError structure.
189 """
190 # Similar to query model caching, cache on the underlying function to support bound methods
191 cache_target = getattr(handler, "__func__", handler)
192 cached_model = _get_cached_combined_model(handler)
193 if cached_model is not None:
194 return cached_model
196 sig = inspect.signature(handler)
197 hints = _cached_get_type_hints_for_handler(handler)
199 # Get explicit metadata if provided via decorators (@query, @header, etc)
200 param_metadata = getattr(cache_target, "_param_metadata", [])
201 metadata_by_pos = {i: m for i, m in enumerate(param_metadata)}
203 fields: dict[str, tuple] = {}
205 for i, (name, param) in enumerate(sig.parameters.items()):
206 if name in ("self", "request"):
207 continue
209 annotation = hints.get(name, param.annotation)
211 if annotation == inspect.Parameter.empty:
212 continue
214 # Determine matching source metadata
215 meta = getattr(param.default, "_lexigram_param_info", None)
216 if meta is None:
217 meta = metadata_by_pos.get(i - 1 if "self" in sig.parameters else i)
219 # Basic default value handling
220 default = param.default if param.default is not inspect.Parameter.empty else ...
222 # Determine if this is strictly a body parameter (Pydantic model or other complex type)
223 is_pydantic = False
224 try:
225 if isinstance(annotation, type) and issubclass(annotation, DomainModel):
226 is_pydantic = True
227 except (TypeError, AttributeError):
228 pass
230 # Also treat plain Pydantic BaseModel subclasses as body params
231 # (zero-annotation auto-injection: `body: CreateUserDTO` without @body decorator)
232 if not is_pydantic:
233 try:
234 from pydantic import BaseModel as _PydanticBM
236 if isinstance(annotation, type) and issubclass(annotation, _PydanticBM):
237 is_pydantic = True
238 except ImportError:
239 pass
241 # If it's a simple type OR explicitly decorated for a source, add it to fields
242 is_explicit_source = meta and meta.get("type") in ("header", "cookie", "form")
243 # If the param has pipes, use Any so pydantic doesn't validate — the pipe does it
244 has_pipes = bool(meta and meta.get("pipes"))
246 if _is_simple_type(annotation) or is_explicit_source:
247 # If default is a Lexigram decorator, use its internal default instead
248 if hasattr(default, "_lexigram_param_info"):
249 decorator_info = default._lexigram_param_info
250 default = decorator_info.get("default")
251 if default is ...:
252 default = ...
254 if meta and meta.get("default") is not ...:
255 default = meta.get("default")
257 fields[name] = (Any if has_pipes else annotation, default)
258 continue
260 # Everything else is treated as a potential body parameter (Pydantic model, dict, list, etc)
261 # unless it was explicitly decorated for something else (handled above).
262 # CRITICAL: We MUST skip parameters that are intended for DI!
263 # If it's a class/type that is not a DomainModel and not a recognized complex type,
264 # we treat it as a DI parameter and exclude it from the Pydantic model to avoid
265 # PydanticSchemaGenerationError for unknown types.
266 is_complex = False
267 origin = get_origin(annotation)
268 if origin in (list, dict, set, tuple, Any) or annotation in (
269 list,
270 dict,
271 set,
272 tuple,
273 Any,
274 ):
275 is_complex = True
277 if is_pydantic or is_complex:
278 fields[name] = (
279 Any if has_pipes else annotation,
280 default if default is not inspect.Parameter.empty else ...,
281 )
282 else:
283 # Likely a DI parameter (e.g. svc: MyService) or something we don't want to validate
284 logger.debug(
285 "Skipping parameter '%s' (%s) from Pydantic model (assumed DI)",
286 name,
287 annotation,
288 )
289 continue
291 if not fields:
292 _set_cached_combined_model(handler, None)
293 return None
295 model_name = f"{getattr(cache_target, '__name__', 'Handler')}_CombinedModel"
296 model = _create_validation_model(model_name, **fields)
297 _set_cached_combined_model(handler, model)
298 return model
301async def validate_and_merge_request(
302 request: Any,
303 handler: Any,
304 path_params: dict[str, Any],
305) -> DomainModel | None:
306 """Validate path, query, and body parameters and return a pydantic model instance.
308 Returns None if there are no fields to validate for this handler.
309 Raises pydantic.ValidationError when validation fails.
310 """
311 model = get_or_build_combined_model(handler)
313 if model is None:
314 return None
316 data: dict[str, Any] = {}
318 # Start with path params and query params (both are dict-like)
319 data.update(path_params or {})
320 data.update(dict(request.query_params))
322 # Pull from headers and cookies if needed
323 cache_target = getattr(handler, "__func__", handler)
324 sig = inspect.signature(handler)
325 param_metadata = getattr(cache_target, "_param_metadata", [])
327 # Map parameter names to metadata (both from defaults and decorators)
328 meta_by_name = {}
329 param_names = list(sig.parameters.keys())
330 offset = 1 if "self" in param_names else 0
332 for i, (p_name, param) in enumerate(sig.parameters.items()):
333 # 1. Check signature default
334 info = getattr(param.default, "_lexigram_param_info", None)
335 if info:
336 meta_by_name[p_name] = info
337 # 2. Check position-based decorator metadata
338 elif i - offset >= 0 and i - offset < len(param_metadata):
339 meta_by_name[p_name] = param_metadata[i - offset]
341 for name, meta in meta_by_name.items():
342 m_type = meta.get("type")
343 lookup_name = meta.get("alias") or meta.get("name") or name
345 if m_type == "header":
346 val = request.headers.get(lookup_name)
347 if val is not None:
348 data[name] = val
349 elif m_type == "cookie":
350 val = request.cookies.get(lookup_name)
351 if val is not None:
352 data[name] = val
354 # Identify fields that sub-select from JSON body or are complex types
355 fields_iter = (
356 getattr(model, "model_fields", None) or getattr(model, "__fields__", None) or {}
357 )
358 if not fields_iter and hasattr(model, "__annotations__"):
359 fields_iter = {k: getattr(model, k, None) for k in model.__annotations__}
361 model_field_names = list(fields_iter.keys())
363 # Body fields are those that are NOT sourced from path, query, header, cookie, or form
364 explicit_sources = ("header", "cookie", "form", "path", "query")
365 body_field_names = []
367 for f_name in model_field_names:
368 meta = meta_by_name.get(f_name)
369 if f_name in (path_params or {}) or f_name in request.query_params:
370 continue
372 if meta and meta.get("type") in explicit_sources:
373 continue
375 # Check annotation - only treat as body if it's NOT a simple type
376 # Get annotation from model field
377 ann = None
378 if hasattr(model, "__annotations__") and f_name in model.__annotations__:
379 ann = model.__annotations__[f_name]
380 else:
381 field_info = fields_iter.get(f_name)
382 if field_info is not None and hasattr(field_info, "annotation"):
383 ann = field_info.annotation
384 elif isinstance(field_info, tuple): # pydantic v1 fallback or tuple
385 ann = field_info[0]
387 if ann and _is_simple_type(ann):
388 continue
390 # It's a candidate for body (complex type or Pydantic model)
391 body_field_names.append(f_name)
393 # Check for form fields too
394 form_field_names = [
395 name for name, meta in meta_by_name.items() if meta.get("type") == "form"
396 ]
398 if body_field_names:
399 body_json = None
400 try:
401 body_json = await request.json()
402 except (ValueError, TypeError) as e:
403 logger.debug("Failed to parse request.json() for body fields: %s", e)
404 body_json = None
406 if isinstance(body_json, dict):
407 for name in body_field_names:
408 # Get the annotation for this body field to check if it's a model
409 ann = None
410 if hasattr(model, "__annotations__") and name in model.__annotations__:
411 ann = model.__annotations__[name]
413 is_model_type = False
414 if ann is not None:
415 try:
416 from pydantic import BaseModel
418 if isinstance(ann, type) and issubclass(ann, BaseModel):
419 is_model_type = True
420 except ImportError:
421 pass
423 if len(body_field_names) == 1 and (
424 name not in body_json or is_model_type
425 ):
426 # Single body field: map whole body to it (handles both
427 # `body: CreateUserDTO` and param name collisions)
428 data[name] = body_json
429 elif name in body_json:
430 data[name] = body_json[name]
431 # Body is not a dict (e.g. list or literal), but we have body fields
432 # If only one body field, map body to it
433 elif len(body_field_names) == 1:
434 data[body_field_names[0]] = body_json
436 if form_field_names:
437 try:
438 form_data = await request.form()
439 for name in form_field_names:
440 meta = meta_by_name[name]
441 lookup_name = meta.get("alias") or meta.get("name") or name
442 val = form_data.get(lookup_name)
443 if val is not None:
444 data[name] = val
445 except (ValueError, TypeError) as e:
446 logger.debug("Failed to parse request.form() for form fields: %s", e)
448 # Let pydantic coerce types and raise ValidationError for any problems
449 try:
450 instance = model(**data)
451 except Exception as e: # noqa: BLE001 — duck-typing on Pydantic ValidationError (v1/v2) requires catching broadly before re-raising
452 from lexigram.contracts.exceptions.domain import (
453 ValidationError as LexigramValidationError,
454 )
456 if isinstance(e, LexigramValidationError):
457 raise
459 # Check if it's a Pydantic ValidationError
460 if type(e).__name__ == "ValidationError":
461 from lexigram.contracts.exceptions.domain import FieldError
463 lex_errors = []
464 # Extract errors from pydantic (works for v1 and v2)
465 pydantic_errors = getattr(e, "errors", None)
466 if callable(pydantic_errors):
467 pydantic_errors = pydantic_errors()
468 elif pydantic_errors is None:
469 pydantic_errors = []
471 for err in pydantic_errors:
472 field = ".".join(str(loc) for loc in err.get("loc", []))
473 lex_errors.append(
474 FieldError(
475 field=field,
476 message=err.get("msg", "Invalid value"),
477 code=err.get("type", "invalid"),
478 )
479 )
480 raise ValidationError("Validation failed", errors=lex_errors) from e
481 raise
483 # Run custom validation callables if present
484 for name, meta in meta_by_name.items():
485 validator = meta.get("validation")
486 if validator and callable(validator):
487 val = getattr(instance, name, None)
488 if val is not None:
489 # If validator returns False or raises, it should be treated as failure
490 try:
491 res = validator(val)
492 if res is False:
493 # Create a mock Lexigram error for consistency
494 from lexigram.contracts.exceptions.domain import FieldError
496 raise ValidationError(
497 "Custom validation failed",
498 errors=[
499 FieldError(
500 field=name,
501 message="Value failed custom validation",
502 code="custom_validation",
503 )
504 ],
505 )
506 except (ValueError, TypeError, AssertionError) as e:
507 from lexigram.contracts.exceptions.domain import FieldError
509 raise ValidationError(
510 "Custom validation failed",
511 errors=[
512 FieldError(
513 field=name,
514 message=str(e),
515 code="custom_validation",
516 )
517 ],
518 ) from e
520 return instance