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

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__( 

157 self, 

158 min: Number | None = None, 

159 max: Number | None = None, 

160 min_exclusive: bool = False, 

161 max_exclusive: bool = False, 

162 ): 

163 """Initialize range constraint. 

164 

165 Args: 

166 min: Minimum value (inclusive by default) 

167 max: Maximum value (inclusive by default) 

168 min_exclusive: If True, minimum is exclusive (value must be > min) 

169 max_exclusive: If True, maximum is exclusive (value must be < max) 

170 """ 

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

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

173 self.min = min 

174 self.max = max 

175 self.min_exclusive = min_exclusive 

176 self.max_exclusive = max_exclusive 

177 

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

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

180 if value is None: 

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

182 

183 if not isinstance(value, Number): 

184 return ValidationResult.failure( 

185 value, 

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

187 ) 

188 

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

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

191 return ValidationResult.failure( 

192 value, 

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

194 ) 

195 

196 errors = [] 

197 if self.min is not None: 

198 if self.min_exclusive: 

199 if float(value) <= float(self.min): # type: ignore[arg-type] 

200 errors.append(f"Value {value} must be greater than {self.min}") 

201 else: 

202 if float(value) < float(self.min): # type: ignore[arg-type] 

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

204 

205 if self.max is not None: 

206 if self.max_exclusive: 

207 if float(value) >= float(self.max): # type: ignore[arg-type] 

208 errors.append(f"Value {value} must be less than {self.max}") 

209 else: 

210 if float(value) > float(self.max): # type: ignore[arg-type] 

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

212 

213 if errors: 

214 return ValidationResult.failure(value, errors) 

215 return ValidationResult.success(value) 

216 

217 

218class Length(Constraint): 

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

220 

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

222 """Initialize length constraint. 

223  

224 Args: 

225 min: Minimum length (inclusive) 

226 max: Maximum length (inclusive) 

227 """ 

228 if min is not None and min < 0: 

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

230 if max is not None and max < 0: 

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

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

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

234 self.min = min 

235 self.max = max 

236 

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

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

239 if value is None: 

240 return ValidationResult.success(value) 

241 

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

243 return ValidationResult.failure( 

244 value, 

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

246 ) 

247 

248 length = len(value) 

249 errors = [] 

250 

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

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

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

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

255 

256 if errors: 

257 return ValidationResult.failure(value, errors) 

258 return ValidationResult.success(value) 

259 

260 

261class Pattern(Constraint): 

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

263 

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

265 """Initialize pattern constraint. 

266  

267 Args: 

268 pattern: Regex pattern (string or compiled pattern) 

269 """ 

270 if isinstance(pattern, str): 

271 self.regex = re.compile(pattern) 

272 else: 

273 self.regex = pattern 

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

275 

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

277 """Check if value matches pattern.""" 

278 if value is None: 

279 return ValidationResult.success(value) 

280 

281 if not isinstance(value, str): 

282 return ValidationResult.failure( 

283 value, 

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

285 ) 

286 

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

288 return ValidationResult.failure( 

289 value, 

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

291 ) 

292 

293 return ValidationResult.success(value) 

294 

295 

296class Enum(Constraint): 

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

298 

299 def __init__(self, values: list[AnyType], case_sensitive: bool = True): 

300 """Initialize enum constraint. 

301 

302 Args: 

303 values: List of allowed values 

304 case_sensitive: If False, string comparisons ignore case 

305 """ 

306 if not values: 

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

308 self.values = values 

309 self.case_sensitive = case_sensitive 

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

311 

312 if case_sensitive: 

313 self.allowed = set(values) 

314 else: 

315 # For case-insensitive, store lowercase versions for comparison 

316 self.allowed_lower = { 

317 v.lower() if isinstance(v, str) else v for v in values 

318 } 

319 

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

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

322 if value is None: 

323 return ValidationResult.success(value) 

324 

325 if self.case_sensitive: 

326 if value not in self.allowed: 

327 return ValidationResult.failure( 

328 value, 

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

330 ) 

331 else: 

332 # Case-insensitive comparison 

333 check_value = value.lower() if isinstance(value, str) else value 

334 if check_value not in self.allowed_lower: 

335 return ValidationResult.failure( 

336 value, 

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

338 ) 

339 

340 return ValidationResult.success(value) 

341 

342 

343class Unique(Constraint): 

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

345 

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

347 """Initialize unique constraint. 

348  

349 Args: 

350 field_name: Optional field name for context tracking 

351 """ 

352 self.field_name = field_name or "default" 

353 

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

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

356 if value is None: 

357 return ValidationResult.success(value) 

358 

359 if context is None: 

360 # Without context, we can't track uniqueness 

361 return ValidationResult.success( 

362 value, 

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

364 ) 

365 

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

367 return ValidationResult.failure( 

368 value, 

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

370 ) 

371 

372 context.mark_seen(self.field_name, value) 

373 return ValidationResult.success(value) 

374 

375 

376class Custom(Constraint): 

377 """Custom constraint using a callable.""" 

378 

379 def __init__( 

380 self, 

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

382 error_message: str = "Custom validation failed" 

383 ): 

384 """Initialize custom constraint. 

385  

386 Args: 

387 validator: Callable that returns bool or ValidationResult 

388 error_message: Error message if validation fails 

389 """ 

390 self.validator = validator 

391 self.error_message = error_message 

392 

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

394 """Check using custom validator.""" 

395 try: 

396 result = self.validator(value) 

397 

398 if isinstance(result, ValidationResult): 

399 return result 

400 elif isinstance(result, bool): 

401 if result: 

402 return ValidationResult.success(value) 

403 else: 

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

405 else: 

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

407 value, 

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

409 ) 

410 except Exception as e: 

411 return ValidationResult.failure( 

412 value, 

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

414 )