Coverage for agentos/tools/data_validator.py: 17%

1""" 

2DataValidator — schema-based data validation with custom rules. 

3 

4Supports: 

5 - Type validation (str, int, float, bool, list, dict) 

6 - Required/optional fields 

7 - Nullable fields 

8 - Min/max for numbers and strings 

9 - Enum (allowed values) 

10 - Regex pattern matching 

11 - Nested object validation 

12 - List item validation 

13 - Custom validator functions 

14 - Human-readable error messages 

15""" 

16 

17from __future__ import annotations 

18 

19import re 

20from collections.abc import Callable 

21from typing import Any 

22 

23# ============================================================================ 

24# Schema definition 

25# ============================================================================ 

26 

27 

28class Field: 

29 """A single field definition within a schema.""" 

30 

31 def __init__( 

32 self, 

33 field_type: type, 

34 required: bool = True, 

35 nullable: bool = False, 

36 min_value: int | float | None = None, 

37 max_value: int | float | None = None, 

38 min_length: int | None = None, 

39 max_length: int | None = None, 

40 enum: list[Any] | None = None, 

41 pattern: str | None = None, 

42 custom: Callable[[Any], str | None] | None = None, 

43 # Nesting 

44 nested: dict[str, Field] | None = None, 

45 items: Field | None = None, 

46 ): 

47 self.field_type = field_type 

48 self.required = required 

49 self.nullable = nullable 

50 self.min_value = min_value 

51 self.max_value = max_value 

52 self.min_length = min_length 

53 self.max_length = max_length 

54 self.enum = enum 

55 self.pattern = re.compile(pattern) if pattern else None 

56 self.custom = custom 

57 self.nested = nested 

58 self.items = items 

59 

60 

61# ============================================================================ 

62# Validator 

63# ============================================================================ 

64 

65 

66class ValidationError(Exception): 

67 """Raised when validation fails. Carries a list of error messages.""" 

68 

69 def __init__(self, errors: list[str]): 

70 self.errors = errors 

71 super().__init__("\n".join(errors)) 

72 

73 

74class DataValidator: 

75 """Schema-based data validator. 

76 

77 Usage: 

78 schema = { 

79 "name": Field(str, min_length=1, max_length=100), 

80 "age": Field(int, min_value=0, max_value=150), 

81 "email": Field(str, pattern=r"^[^@]+@[^@]+\\.[^@]+$"), 

82 "tags": Field(list, items=Field(str)), 

83 } 

84 

85 validator = DataValidator(schema) 

86 result = validator.validate(data) 

87 if result: 

88 ... # use result 

89 """ 

90 

91 def __init__(self, schema: dict[str, Field]): 

92 self._schema = schema 

93 

94 def validate(self, data: dict) -> dict: 

95 """Validate data against schema. Returns cleaned data or raises ValidationError.""" 

96 errors = [] 

97 cleaned = self._validate_dict(data, self._schema, "", errors) 

98 if errors: 

99 raise ValidationError(errors) 

100 return cleaned 

101 

102 def is_valid(self, data: dict) -> bool: 

103 """Check if data is valid without raising.""" 

104 try: 

105 self.validate(data) 

106 return True 

107 except ValidationError: 

108 return False 

109 

110 def errors(self, data: dict) -> list[str]: 

111 """Return list of validation error messages.""" 

112 errors_list: list[str] = [] 

113 self._validate_dict(data, self._schema, "", errors_list) 

114 return errors_list 

115 

116 # ---------- Internal ---------- 

117 

118 def _validate_dict( 

119 self, data: dict, schema: dict[str, Field], path: str, errors: list[str] 

120 ) -> dict: 

121 if not isinstance(data, dict): 

122 errors.append(f"{path or '(root)'}: expected dict, got {type(data).__name__}") 

123 return {} 

124 

125 result = {} 

126 

127 # Check required fields 

128 for name, field in schema.items(): 

129 fpath = f"{path}.{name}" if path else name 

130 if name not in data: 

131 if field.required: 

132 errors.append(f"{fpath}: required field missing") 

133 continue 

134 

135 value = data[name] 

136 validated = self._validate_value(value, field, fpath, errors) 

137 if validated is not None or field.nullable: 

138 result[name] = validated 

139 

140 # Warn about unknown fields (can be made strict later) 

141 return result 

142 

143 def _validate_value(self, value: Any, field: Field, path: str, errors: list[str]) -> Any: 

144 # Nullable check 

145 if value is None: 

146 if not field.nullable: 

147 errors.append(f"{path}: value is None but field is not nullable") 

148 return None 

149 return None 

150 

151 # Type check 

152 if not isinstance(value, field.field_type): 

153 errors.append( 

154 f"{path}: expected {field.field_type.__name__}, got {type(value).__name__}" 

155 ) 

156 return None 

157 

158 # Min/max for numbers 

159 if field.field_type in (int, float): 

160 if field.min_value is not None and value < field.min_value: 

161 errors.append(f"{path}: value {value} < min {field.min_value}") 

162 if field.max_value is not None and value > field.max_value: 

163 errors.append(f"{path}: value {value} > max {field.max_value}") 

164 

165 # Length for strings 

166 if field.field_type is str: 

167 if field.min_length is not None and len(value) < field.min_length: 

168 errors.append(f"{path}: length {len(value)} < min {field.min_length}") 

169 if field.max_length is not None and len(value) > field.max_length: 

170 errors.append(f"{path}: length {len(value)} > max {field.max_length}") 

171 

172 # Enum 

173 if field.enum is not None and value not in field.enum: 

174 errors.append(f"{path}: {value!r} not in {field.enum}") 

175 

176 # Pattern (regex) 

177 if field.pattern and not field.pattern.search(str(value)): 

178 errors.append(f"{path}: {value!r} does not match pattern") 

179 

180 # Nested object 

181 if field.nested and isinstance(value, dict): 

182 value = self._validate_dict(value, field.nested, path, errors) 

183 

184 # List items 

185 if field.items and isinstance(value, list): 

186 value = self._validate_list(value, field.items, path, errors) 

187 

188 # Custom validator 

189 if field.custom: 

190 msg = field.custom(value) 

191 if msg: 

192 errors.append(f"{path}: {msg}") 

193 

194 return value 

195 

196 def _validate_list(self, data: list, item_field: Field, path: str, errors: list[str]) -> list: 

197 result = [] 

198 for i, item in enumerate(data): 

199 item_path = f"{path}[{i}]" 

200 validated = self._validate_value(item, item_field, item_path, errors) 

201 result.append(validated) 

202 return result