Coverage for src \ ingress_recipient \ validation.py: 100%

1"""Validation utilities for recipient data.""" 

2 

3from typing import Any 

4 

5from pydantic import ValidationError 

6 

7from .types import RecipientData 

8 

9 

10def validate_recipient_data(data: dict[str, Any]) -> RecipientData: 

11 """Validate recipient data dictionary. 

12 

13 Parses and validates a dictionary against the RecipientData model. 

14 

15 Args: 

16 data: Dictionary to validate 

17 

18 Returns: 

19 Validated RecipientData object 

20 

21 Raises: 

22 ValidationError: If validation fails 

23 

24 Example: 

25 ```python 

26 try: 

27 validated = validate_recipient_data({ 

28 "channel_id": "ch_test", 

29 "tenant_id": "123e4567-e89b-12d3-a456-426614174000", 

30 "timestamp": "2025-12-15T12:00:00Z", 

31 "data": {"value": 1}, 

32 "metadata": {"ingress_timestamp": "2025-12-15T12:00:01Z"}, 

33 }) 

34 # Use validated data... 

35 except ValidationError as e: 

36 print(f"Validation failed: {e}") 

37 ``` 

38 """ 

39 return RecipientData.model_validate(data) 

40 

41 

42def safe_validate_recipient_data( 

43 data: dict[str, Any], 

44) -> tuple[RecipientData | None, list[dict[str, Any]] | None]: 

45 """Safely validate recipient data without raising exceptions. 

46 

47 Returns a tuple of (validated_data, errors). Exactly one will be None. 

48 

49 Args: 

50 data: Dictionary to validate 

51 

52 Returns: 

53 Tuple of (validated_data, errors): 

54 - If validation succeeds: (RecipientData, None) 

55 - If validation fails: (None, list of error dictionaries) 

56 

57 Example: 

58 ```python 

59 validated, errors = safe_validate_recipient_data(data) 

60 

61 if validated: 

62 # Use validated data 

63 print(f"Channel: {validated.channel_id}") 

64 else: 

65 # Handle errors 

66 for error in errors: 

67 print(f"Error at {error['loc']}: {error['msg']}") 

68 ``` 

69 """ 

70 try: 

71 validated = RecipientData.model_validate(data) 

72 return validated, None 

73 except ValidationError as e: 

74 errors = [ 

75 { 

76 "loc": list(error["loc"]), 

77 "msg": error["msg"], 

78 "type": error["type"], 

79 } 

80 for error in e.errors() 

81 ] 

82 return None, errors