Coverage for src/lexigram/web/pipes/builtin/validation.py: 17%

1"""Validation pipe for Pydantic model validation. 

2 

3Validates incoming data against Pydantic models. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from lexigram.web.protocols import ParamMetadata, PipeProtocol 

11 

12 

13class ValidationPipe(PipeProtocol): 

14 """PipeProtocol that validates data against Pydantic models. 

15 

16 Uses Pydantic's model_validate() for validation. 

17 

18 Example: 

19 ```python 

20 from lexigram.domain import DomainModel 

21 

22 @dataclass(init=False) 

23 class CreateUserRequest(DomainModel): 

24 name: str 

25 email: str 

26 

27 class UserController(Controller): 

28 @post("/users") 

29 async def create_user(self, @body(pipe=ValidationPipe()) data: CreateUserRequest): 

30 ... 

31 ``` 

32 """ 

33 

34 def __init__(self, model: type | None = None): 

35 """Initialize the validation pipe. 

36 

37 Args: 

38 model: Optional Pydantic model class to validate against. 

39 """ 

40 self._model = model 

41 

42 async def transform(self, value: Any, metadata: ParamMetadata) -> Any: 

43 """Validate the value against the Pydantic model. 

44 

45 Args: 

46 value: The value to validate. 

47 metadata: Metadata about the parameter. 

48 

49 Returns: 

50 Validated model instance. 

51 

52 Raises: 

53 ValidationError: If validation fails. 

54 """ 

55 from lexigram.contracts.exceptions.domain import ValidationError 

56 

57 if value is None: 

58 return None 

59 

60 # Use provided model or try to infer from metadata 

61 model = self._model 

62 

63 if model is None: 

64 # Try to get expected type from metadata 

65 expected_type = metadata.expected_type 

66 if expected_type and hasattr(expected_type, "model_validate"): 

67 model = expected_type 

68 

69 if model is None: 

70 # No model to validate against 

71 return value 

72 

73 try: 

74 if hasattr(model, "model_validate"): 

75 # Pydantic v2 

76 return model.model_validate(value) 

77 if hasattr(model, "parse_obj"): 

78 # Pydantic v1 

79 return model.parse_obj(value) 

80 # Not a Pydantic model 

81 return value 

82 except Exception as e: # noqa: BLE001 — duck-typing on Pydantic ValidationError (v1/v2) requires catching broadly before re-raising 

83 from lexigram.contracts.exceptions.domain import ( 

84 ValidationError as LexigramValidationError, 

85 ) 

86 

87 if isinstance(e, LexigramValidationError): 

88 raise 

89 

90 # Check if it's a Pydantic ValidationError 

91 if type(e).__name__ == "ValidationError": 

92 from lexigram.contracts.exceptions.domain import FieldError 

93 

94 lex_errors = [] 

95 # Extract errors from pydantic (works for v1 and v2) 

96 pydantic_errors = getattr(e, "errors", None) 

97 if callable(pydantic_errors): 

98 pydantic_errors = pydantic_errors() 

99 elif pydantic_errors is None: 

100 pydantic_errors = [] 

101 

102 for err in pydantic_errors: 

103 # For a single parameter pipe, the field name is usually the parameter name 

104 # but we can try to get it from location 

105 loc = err.get("loc", []) 

106 field = ( 

107 ".".join(str(part) for part in loc) if loc else metadata.name 

108 ) 

109 

110 lex_errors.append( 

111 FieldError( 

112 field=field, 

113 message=err.get("msg", "Invalid value"), 

114 code=err.get("type", "invalid"), 

115 ) 

116 ) 

117 raise ValidationError( 

118 f"Validation failed for {metadata.name}", errors=lex_errors 

119 ) from e 

120 raise 

121 

122 

123__all__ = ["ValidationPipe"]