Coverage for src/dataknobs_data/validation_v2/constraints.py: 89%

1""" 

2Constraint implementations with consistent, composable API. 

3""" 

4 

5import math 

6import re 

7from abc import ABC, abstractmethod 

8from typing import Any, List, Optional, Callable, Union, Pattern as RegexPattern 

9from numbers import Number 

10 

11from .result import ValidationResult, ValidationContext 

12 

13 

14class Constraint(ABC): 

15 """Base class for all constraints with composable operators.""" 

16 

17 @abstractmethod 

18 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

19 """ 

20 Validate a value against this constraint. 

21  

22 Args: 

23 value: Value to validate 

24 context: Optional validation context for stateful constraints 

25  

26 Returns: 

27 ValidationResult with validation outcome 

28 """ 

29 pass 

30 

31 def __and__(self, other: 'Constraint') -> 'All': 

32 """Combine with AND: both constraints must pass.""" 

33 if isinstance(self, All): 

34 return All(self.constraints + [other]) 

35 elif isinstance(other, All): 

36 return All([self] + other.constraints) 

37 return All([self, other]) 

38 

39 def __or__(self, other: 'Constraint') -> 'Any': 

40 """Combine with OR: at least one constraint must pass.""" 

41 if isinstance(self, Any): 

42 return Any(self.constraints + [other]) 

43 elif isinstance(other, Any): 

44 return Any([self] + other.constraints) 

45 return Any([self, other]) 

46 

47 def __invert__(self) -> 'Not': 

48 """Negate this constraint.""" 

49 return Not(self) 

50 

51 

52class All(Constraint): 

53 """All constraints must pass (AND logic).""" 

54 

55 def __init__(self, constraints: List[Constraint]): 

56 """ 

57 Initialize with list of constraints. 

58  

59 Args: 

60 constraints: List of constraints that must all pass 

61 """ 

62 self.constraints = constraints 

63 

64 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

65 """Check all constraints.""" 

66 result = ValidationResult.success(value) 

67 

68 for constraint in self.constraints: 

69 check_result = constraint.check(value, context) 

70 if not check_result.valid: 

71 result = result.merge(check_result) 

72 # Continue checking to collect all errors 

73 

74 return result 

75 

76 

77class Any(Constraint): 

78 """At least one constraint must pass (OR logic).""" 

79 

80 def __init__(self, constraints: List[Constraint]): 

81 """ 

82 Initialize with list of constraints. 

83  

84 Args: 

85 constraints: List of constraints where at least one must pass 

86 """ 

87 self.constraints = constraints 

88 

89 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

90 """Check if any constraint passes.""" 

91 all_errors = [] 

92 

93 for constraint in self.constraints: 

94 check_result = constraint.check(value, context) 

95 if check_result.valid: 

96 return check_result 

97 all_errors.extend(check_result.errors) 

98 

99 return ValidationResult.failure( 

100 value, 

101 [f"None of the constraints passed: {', '.join(all_errors)}"] 

102 ) 

103 

104 

105class Not(Constraint): 

106 """Negates a constraint.""" 

107 

108 def __init__(self, constraint: Constraint): 

109 """ 

110 Initialize with constraint to negate. 

111  

112 Args: 

113 constraint: Constraint to negate 

114 """ 

115 self.constraint = constraint 

116 

117 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

118 """Check if constraint fails (negation).""" 

119 result = self.constraint.check(value, context) 

120 if result.valid: 

121 return ValidationResult.failure( 

122 value, 

123 [f"Value should not satisfy constraint but it does"] 

124 ) 

125 return ValidationResult.success(value) 

126 

127 

128class Required(Constraint): 

129 """Field must be present and non-null.""" 

130 

131 def __init__(self, allow_empty: bool = False): 

132 """ 

133 Initialize required constraint. 

134  

135 Args: 

136 allow_empty: If True, empty strings/collections are allowed 

137 """ 

138 self.allow_empty = allow_empty 

139 

140 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

141 """Check if value is present and non-null.""" 

142 if value is None: 

143 return ValidationResult.failure(value, ["Value is required"]) 

144 

145 if not self.allow_empty: 

146 # Check for empty strings and collections 

147 if isinstance(value, (str, list, dict, set, tuple)) and len(value) == 0: 

148 return ValidationResult.failure(value, ["Value cannot be empty"]) 

149 

150 return ValidationResult.success(value) 

151 

152 

153class Range(Constraint): 

154 """Numeric value must be in specified range.""" 

155 

156 def __init__(self, min: Optional[Number] = None, max: Optional[Number] = None): 

157 """ 

158 Initialize range constraint. 

159  

160 Args: 

161 min: Minimum value (inclusive) 

162 max: Maximum value (inclusive) 

163 """ 

164 if min is not None and max is not None and min > max: 

165 raise ValueError(f"min ({min}) cannot be greater than max ({max})") 

166 self.min = min 

167 self.max = max 

168 

169 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

170 """Check if value is in range.""" 

171 if value is None: 

172 return ValidationResult.success(value) # None is considered valid (use Required to enforce) 

173 

174 if not isinstance(value, Number): 

175 return ValidationResult.failure( 

176 value, 

177 [f"Value must be a number, got {type(value).__name__}"] 

178 ) 

179 

180 # Check for NaN (Not a Number) - NaN is not valid for range comparisons 

181 if isinstance(value, float) and math.isnan(value): 

182 return ValidationResult.failure( 

183 value, 

184 ["Value is NaN (Not a Number), which is not valid for range comparisons"] 

185 ) 

186 

187 errors = [] 

188 if self.min is not None and value < self.min: 

189 errors.append(f"Value {value} is less than minimum {self.min}") 

190 if self.max is not None and value > self.max: 

191 errors.append(f"Value {value} is greater than maximum {self.max}") 

192 

193 if errors: 

194 return ValidationResult.failure(value, errors) 

195 return ValidationResult.success(value) 

196 

197 

198class Length(Constraint): 

199 """String/collection length must be in specified range.""" 

200 

201 def __init__(self, min: Optional[int] = None, max: Optional[int] = None): 

202 """ 

203 Initialize length constraint. 

204  

205 Args: 

206 min: Minimum length (inclusive) 

207 max: Maximum length (inclusive) 

208 """ 

209 if min is not None and min < 0: 

210 raise ValueError(f"min length cannot be negative: {min}") 

211 if max is not None and max < 0: 

212 raise ValueError(f"max length cannot be negative: {max}") 

213 if min is not None and max is not None and min > max: 

214 raise ValueError(f"min length ({min}) cannot be greater than max ({max})") 

215 self.min = min 

216 self.max = max 

217 

218 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

219 """Check if value length is in range.""" 

220 if value is None: 

221 return ValidationResult.success(value) 

222 

223 if not hasattr(value, '__len__'): 

224 return ValidationResult.failure( 

225 value, 

226 [f"Value does not have a length: {type(value).__name__}"] 

227 ) 

228 

229 length = len(value) 

230 errors = [] 

231 

232 if self.min is not None and length < self.min: 

233 errors.append(f"Length {length} is less than minimum {self.min}") 

234 if self.max is not None and length > self.max: 

235 errors.append(f"Length {length} is greater than maximum {self.max}") 

236 

237 if errors: 

238 return ValidationResult.failure(value, errors) 

239 return ValidationResult.success(value) 

240 

241 

242class Pattern(Constraint): 

243 """String value must match regex pattern.""" 

244 

245 def __init__(self, pattern: Union[str, RegexPattern]): 

246 """ 

247 Initialize pattern constraint. 

248  

249 Args: 

250 pattern: Regex pattern (string or compiled pattern) 

251 """ 

252 if isinstance(pattern, str): 

253 self.regex = re.compile(pattern) 

254 else: 

255 self.regex = pattern 

256 self.pattern_str = pattern if isinstance(pattern, str) else pattern.pattern 

257 

258 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

259 """Check if value matches pattern.""" 

260 if value is None: 

261 return ValidationResult.success(value) 

262 

263 if not isinstance(value, str): 

264 return ValidationResult.failure( 

265 value, 

266 [f"Value must be a string for pattern matching, got {type(value).__name__}"] 

267 ) 

268 

269 if not self.regex.match(value): 

270 return ValidationResult.failure( 

271 value, 

272 [f"Value '{value}' does not match pattern '{self.pattern_str}'"] 

273 ) 

274 

275 return ValidationResult.success(value) 

276 

277 

278class Enum(Constraint): 

279 """Value must be in allowed set.""" 

280 

281 def __init__(self, values: List[Any]): 

282 """ 

283 Initialize enum constraint. 

284  

285 Args: 

286 values: List of allowed values 

287 """ 

288 if not values: 

289 raise ValueError("Enum constraint requires at least one allowed value") 

290 self.allowed = set(values) 

291 self.allowed_str = ', '.join(repr(v) for v in values) 

292 

293 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

294 """Check if value is in allowed set.""" 

295 if value is None: 

296 return ValidationResult.success(value) 

297 

298 if value not in self.allowed: 

299 return ValidationResult.failure( 

300 value, 

301 [f"Value '{value}' is not in allowed values: {self.allowed_str}"] 

302 ) 

303 

304 return ValidationResult.success(value) 

305 

306 

307class Unique(Constraint): 

308 """Value must be unique (uses context for tracking).""" 

309 

310 def __init__(self, field_name: Optional[str] = None): 

311 """ 

312 Initialize unique constraint. 

313  

314 Args: 

315 field_name: Optional field name for context tracking 

316 """ 

317 self.field_name = field_name or "default" 

318 

319 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

320 """Check if value is unique using context.""" 

321 if value is None: 

322 return ValidationResult.success(value) 

323 

324 if context is None: 

325 # Without context, we can't track uniqueness 

326 return ValidationResult.success( 

327 value, 

328 warnings=["Unique constraint requires context for tracking"] 

329 ) 

330 

331 if context.has_seen(self.field_name, value): 

332 return ValidationResult.failure( 

333 value, 

334 [f"Duplicate value '{value}' for field '{self.field_name}'"] 

335 ) 

336 

337 context.mark_seen(self.field_name, value) 

338 return ValidationResult.success(value) 

339 

340 

341class Custom(Constraint): 

342 """Custom constraint using a callable.""" 

343 

344 def __init__( 

345 self, 

346 validator: Callable[[Any], Union[bool, ValidationResult]], 

347 error_message: str = "Custom validation failed" 

348 ): 

349 """ 

350 Initialize custom constraint. 

351  

352 Args: 

353 validator: Callable that returns bool or ValidationResult 

354 error_message: Error message if validation fails 

355 """ 

356 self.validator = validator 

357 self.error_message = error_message 

358 

359 def check(self, value: Any, context: Optional[ValidationContext] = None) -> ValidationResult: 

360 """Check using custom validator.""" 

361 try: 

362 result = self.validator(value) 

363 

364 if isinstance(result, ValidationResult): 

365 return result 

366 elif isinstance(result, bool): 

367 if result: 

368 return ValidationResult.success(value) 

369 else: 

370 return ValidationResult.failure(value, [self.error_message]) 

371 else: 

372 return ValidationResult.failure( 

373 value, 

374 [f"Custom validator returned unexpected type: {type(result).__name__}"] 

375 ) 

376 except Exception as e: 

377 return ValidationResult.failure( 

378 value, 

379 [f"Custom validation error: {str(e)}"] 

380 )