Coverage for src/dataknobs_data/validation/constraints.py: 26%

1"""Constraint implementations with consistent, composable API. 

2""" 

3 

4from __future__ import annotations 

5 

6import math 

7import re 

8from abc import ABC, abstractmethod 

9from numbers import Number 

10from re import Pattern as RegexPattern 

11from typing import Any as AnyType, TYPE_CHECKING 

12 

13from .result import ValidationContext, ValidationResult 

14 

15if TYPE_CHECKING: 

16 from collections.abc import Callable 

17 

18 

19class Constraint(ABC): 

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

21 

22 @abstractmethod 

23 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

24 """Validate a value against this constraint. 

25  

26 Args: 

27 value: Value to validate 

28 context: Optional validation context for stateful constraints 

29  

30 Returns: 

31 ValidationResult with validation outcome 

32 """ 

33 pass 

34 

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

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

37 if isinstance(self, All): 

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

39 elif isinstance(other, All): 

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

41 return All([self, other]) 

42 

43 def __or__(self, other: Constraint) -> AnyOf: 

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

45 if isinstance(self, AnyOf): 

46 return AnyOf(self.constraints + [other]) 

47 elif isinstance(other, AnyOf): 

48 return AnyOf([self] + other.constraints) 

49 return AnyOf([self, other]) 

50 

51 def __invert__(self) -> Not: 

52 """Negate this constraint.""" 

53 return Not(self) 

54 

55 

56class All(Constraint): 

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

58 

59 def __init__(self, constraints: list[Constraint]): 

60 """Initialize with list of constraints. 

61  

62 Args: 

63 constraints: List of constraints that must all pass 

64 """ 

65 self.constraints = constraints 

66 

67 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

68 """Check all constraints.""" 

69 result = ValidationResult.success(value) 

70 

71 for constraint in self.constraints: 

72 check_result = constraint.check(value, context) 

73 if not check_result.valid: 

74 result = result.merge(check_result) 

75 # Continue checking to collect all errors 

76 

77 return result 

78 

79 

80class AnyOf(Constraint): 

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

82 

83 def __init__(self, constraints: list[Constraint]): 

84 """Initialize with list of constraints. 

85  

86 Args: 

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

88 """ 

89 self.constraints = constraints 

90 

91 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

92 """Check if any constraint passes.""" 

93 all_errors = [] 

94 

95 for constraint in self.constraints: 

96 check_result = constraint.check(value, context) 

97 if check_result.valid: 

98 return check_result 

99 all_errors.extend(check_result.errors) 

100 

101 return ValidationResult.failure( 

102 value, 

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

104 ) 

105 

106 

107class Not(Constraint): 

108 """Negates a constraint.""" 

109 

110 def __init__(self, constraint: Constraint): 

111 """Initialize with constraint to negate. 

112  

113 Args: 

114 constraint: Constraint to negate 

115 """ 

116 self.constraint = constraint 

117 

118 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

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

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

121 if result.valid: 

122 return ValidationResult.failure( 

123 value, 

124 ["Value should not satisfy constraint but it does"] 

125 ) 

126 return ValidationResult.success(value) 

127 

128 

129class Required(Constraint): 

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

131 

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

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: AnyType, context: ValidationContext | None = 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: Number | None = None, max: Number | None = None): 

157 """Initialize range constraint. 

158  

159 Args: 

160 min: Minimum value (inclusive) 

161 max: Maximum value (inclusive) 

162 """ 

163 if min is not None and max is not None and float(min) > float(max): # type: ignore[arg-type] 

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

165 self.min = min 

166 self.max = max 

167 

168 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

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

170 if value is None: 

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

172 

173 if not isinstance(value, Number): 

174 return ValidationResult.failure( 

175 value, 

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

177 ) 

178 

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

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

181 return ValidationResult.failure( 

182 value, 

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

184 ) 

185 

186 errors = [] 

187 if self.min is not None and float(value) < float(self.min): # type: ignore[arg-type] 

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

189 if self.max is not None and float(value) > float(self.max): # type: ignore[arg-type] 

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

191 

192 if errors: 

193 return ValidationResult.failure(value, errors) 

194 return ValidationResult.success(value) 

195 

196 

197class Length(Constraint): 

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

199 

200 def __init__(self, min: int | None = None, max: int | None = None): 

201 """Initialize length constraint. 

202  

203 Args: 

204 min: Minimum length (inclusive) 

205 max: Maximum length (inclusive) 

206 """ 

207 if min is not None and min < 0: 

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

209 if max is not None and max < 0: 

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

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

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

213 self.min = min 

214 self.max = max 

215 

216 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

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

218 if value is None: 

219 return ValidationResult.success(value) 

220 

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

222 return ValidationResult.failure( 

223 value, 

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

225 ) 

226 

227 length = len(value) 

228 errors = [] 

229 

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

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

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

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

234 

235 if errors: 

236 return ValidationResult.failure(value, errors) 

237 return ValidationResult.success(value) 

238 

239 

240class Pattern(Constraint): 

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

242 

243 def __init__(self, pattern: str | RegexPattern): 

244 """Initialize pattern constraint. 

245  

246 Args: 

247 pattern: Regex pattern (string or compiled pattern) 

248 """ 

249 if isinstance(pattern, str): 

250 self.regex = re.compile(pattern) 

251 else: 

252 self.regex = pattern 

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

254 

255 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

256 """Check if value matches pattern.""" 

257 if value is None: 

258 return ValidationResult.success(value) 

259 

260 if not isinstance(value, str): 

261 return ValidationResult.failure( 

262 value, 

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

264 ) 

265 

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

267 return ValidationResult.failure( 

268 value, 

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

270 ) 

271 

272 return ValidationResult.success(value) 

273 

274 

275class Enum(Constraint): 

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

277 

278 def __init__(self, values: list[AnyType]): 

279 """Initialize enum constraint. 

280  

281 Args: 

282 values: List of allowed values 

283 """ 

284 if not values: 

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

286 self.allowed = set(values) 

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

288 

289 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

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

291 if value is None: 

292 return ValidationResult.success(value) 

293 

294 if value not in self.allowed: 

295 return ValidationResult.failure( 

296 value, 

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

298 ) 

299 

300 return ValidationResult.success(value) 

301 

302 

303class Unique(Constraint): 

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

305 

306 def __init__(self, field_name: str | None = None): 

307 """Initialize unique constraint. 

308  

309 Args: 

310 field_name: Optional field name for context tracking 

311 """ 

312 self.field_name = field_name or "default" 

313 

314 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

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

316 if value is None: 

317 return ValidationResult.success(value) 

318 

319 if context is None: 

320 # Without context, we can't track uniqueness 

321 return ValidationResult.success( 

322 value, 

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

324 ) 

325 

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

327 return ValidationResult.failure( 

328 value, 

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

330 ) 

331 

332 context.mark_seen(self.field_name, value) 

333 return ValidationResult.success(value) 

334 

335 

336class Custom(Constraint): 

337 """Custom constraint using a callable.""" 

338 

339 def __init__( 

340 self, 

341 validator: Callable[[AnyType], bool | ValidationResult], 

342 error_message: str = "Custom validation failed" 

343 ): 

344 """Initialize custom constraint. 

345  

346 Args: 

347 validator: Callable that returns bool or ValidationResult 

348 error_message: Error message if validation fails 

349 """ 

350 self.validator = validator 

351 self.error_message = error_message 

352 

353 def check(self, value: AnyType, context: ValidationContext | None = None) -> ValidationResult: 

354 """Check using custom validator.""" 

355 try: 

356 result = self.validator(value) 

357 

358 if isinstance(result, ValidationResult): 

359 return result 

360 elif isinstance(result, bool): 

361 if result: 

362 return ValidationResult.success(value) 

363 else: 

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

365 else: 

366 return ValidationResult.failure( # type: ignore[unreachable] 

367 value, 

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

369 ) 

370 except Exception as e: 

371 return ValidationResult.failure( 

372 value, 

373 [f"Custom validation error: {e!s}"] 

374 )