constrained_values
A Python library for creating type-safe, self-validating value objects using a powerful transformation and validation pipeline.
The Philosophy: Beyond Primitive Types
In many applications, especially when interacting with hardware or external systems, we often pass around primitive types like integers, strings, or floats. This can lead to problems:
- Primitive Obsession: Is
temperature = 25in Celsius or Fahrenheit? Isspi_mode = 2a valid mode? Raw values lack context and safety. - Lost Domain Knowledge: The rules governing these values are scattered throughout the codebase. An
Ageshouldn't be negative, and aTemperaturefrom a sensor might have a specific valid range. - Bugs and Unreliability: Passing an invalid value can lead to subtle bugs or crashes far from where the value was created.
The Constrained Values library solves this by embracing Object-Oriented principles. Instead of passing around a raw int, you create a rich, meaningful Age or Temperature object. This object encapsulates not just the value, but also the rules that govern it, ensuring that it can never exist in an invalid state.
This is particularly powerful for abstracting hardware domains. Instead of remembering that a Modbus (Ventilation Comms) register value of -32768 on a specific hardware means "no sensor detected," or that a valid Serial Peripheral Interface (SPI) "mode" is an integer between 0 and 3, you can create type-safe objects like VentilationTemperature or SPIMode that handle this complexity internally.
Features
- Create Rich Value Objects: Turn primitive data into meaningful, type-safe objects.
- Powerful Validation Pipelines: Chain multiple validation and transformation steps.
- Built-in Strategies: Includes common validators for ranges, enums, types, and more.
- Custom Logic: Easily extend the library with your own validation and transformation strategies.
- Clear Error Handling: Each constrained value clearly reports its status (
OKorEXCEPTION) and provides descriptive error messages. - Optional Error Throw: When constructing a constrained value you can make it throw immediately, so you know an object is valid.
- Type Safety: Enforces the final, canonical type of your value.
Typed, Pipeline-Driven Value Objects
This module provides a small set of primitives for building typed, pipeline-driven value objects:
Value[T]β An immutable, typed wrapper that defines equality and ordering
only against the same concrete subclass.ValidationStrategyβ A pluggable check that returns aStatusResponse
(Status.OKorStatus.EXCEPTION) without changing the value.TransformationStrategyβ A pluggable step that converts an input value of typeGeneric[InT]to
a new value and returns aResponse[OutT](status, details, value).ConstrainedValue[T]β A value object that runs an input through a sequence
of strategies (transformations and validations) to produce a canonicalT,
plus status and details.
Typical Flow
- Start from a raw input (which may not yet be of type
T). - Thread it through a pipeline: transformations may change the value or type,
while validations only inspect it. - On the first
Status.EXCEPTION, the pipeline short-circuits. Otherwise,
the final value is accepted and exposed as the canonicalT.
Quickstart
Let's explore the library's features, starting with a simple case and building up to a complex real-world scenario.
Example 1: Simple Range Validation
The most basic use case is ensuring a value falls within a specific range. Instead of passing an integer around and checking its bounds everywhere, we create an Age type by defining a class.
# See: examples/example_1.py
# Define an 'Age' type that must be an integer between 0 and 120.
class Age(RangeValue[int]):
def __init__(self, value):
super().__init__(value, 10, 120)
# Now, let's use our new Age type.
valid_age = Age(30)
invalid_age = Age(150)
invalid_age_by_type = Age("21")
another_valid_age = Age(32)
print(
f"Another valid age: {another_valid_age.value}, "
f"is greater than valid age: {valid_age.value} ? "
f"{another_valid_age > valid_age}"
)
print(f"Valid age: {valid_age.value}, Is OK: {valid_age.ok}")
print(f"Invalid age: {invalid_age.value}, Is OK: {invalid_age.ok}")
print(f"Error details: {invalid_age.details}")
print(f"Error details: {invalid_age_by_type.details}")
Output
Another valid age: 32, is greater than valid age: 30 ? True
Valid age: 30, Is OK: True
Invalid age: None, Is OK: False
Error details: Value must be less than or equal to 120, got 150
Error details: Value must be one of 'int', got 'str'
Example 2: Simple Range Validation
Example: Using RangeValue with a custom transform (Fahrenheit β Celsius).
This demo shows how to subclass RangeValue and override
get_custom_strategies() to insert a transformation step into the pipeline.
- Input values are provided in Fahrenheit (int or float).
- A FahrenheitToCelsius transformation converts them to Celsius.
- The resulting Celsius values are validated against a range of -10Β°C .. 40Β°C.
- Results are rounded to two decimal places.
Note: The RangeValue class automatically infers acceptable numeric types from your bounds. For example, if your bounds are floats, both int and float inputs are accepted and coerced to float. This inference is handled internally by RangeValue.infer_valid_types_from_value().
# See: examples/example_2.py
class FahrenheitToCelsius(TransformationStrategy[float, float]):
"""
Define a transformation strategy for Fahrenheit.
input and output types are float
"""
def transform(self, value: float) -> Response[float]:
try:
c = round((float(value) - 32.0) * (5.0 / 9.0), 2)
return Response(Status.OK, DEFAULT_SUCCESS_MESSAGE, c)
except Exception as e:
return Response(Status.EXCEPTION, str(e), None)
class FahrenheitToCelsiusValue(RangeValue[float]):
"""
Valid Celsius value between -10 and 40, inclusive.
Accepts input as Fahrenheit (int/float).
Fahrenheit is converted internally to Celsius before validation.
"""
def __init__(self, value: int | float):
super().__init__(value, -10.0, 40.0)
def get_custom_strategies(self):
return [FahrenheitToCelsius()]
print("\n=== Fahrenheit inputs (converted to Celsius) ===")
for val in [50, 50.36, 72]:
cv = FahrenheitToCelsiusValue(val)
print(f"Input {val!r}F β status={cv.status}, value={cv.value}Β°C")
print("\n=== Out of range examples ===")
for val in [-40, 10, 122]:
cv = FahrenheitToCelsiusValue(val)
print(f"Input {val!r} β status={cv.status}, \n details={cv.details}")
Output
=== Fahrenheit inputs (converted to Celsius) ===
Input 50F β status=Status.OK, value=10.0Β°C
Input 50.36F β status=Status.OK, value=10.2Β°C
Input 72F β status=Status.OK, value=22.22Β°C
=== Out of range examples ===
Input -40 β status=Status.EXCEPTION,
details=Value must be greater than or equal to -10.0, got -40.0
Input 10 β status=Status.EXCEPTION,
details=Value must be greater than or equal to -10.0, got -12.22
Input 122 β status=Status.EXCEPTION,
details=Value must be less than or equal to 40.0, got 50.0
Example 3: Complex Pipelines for Hardware Data
This is where the library truly shines. Let's model a real-world hardware scenario: reading a temperature from a ventilation unit via the Modbus protocol.
The process involves multiple steps:
- The input is a register address (an
int). - We must validate that we are allowed to read from this register.
- We fetch the raw integer value from a list of all Modbus registers.
- The hardware uses special values (
-32768,32767) to signal errors like a missing or short-circuited sensor. We must detect these. - If the value is valid, it's not yet in Celsius. We need to divide it by
10.0to get the final temperature.
Hereβs how you can model this entire chain of validation and transformation using a custom ConstrainedRangeValue.
# See: examples/example_3.py
class AllowedInputRegister(ValidationStrategy[int]):
"""Checks if the selected register address is valid."""
def validate(self, value: int) -> StatusResponse:
valid_registers = {0, 1, 2, 3}
if value in valid_registers:
return StatusResponse(status=Status.OK,
details=DEFAULT_SUCCESS_MESSAGE)
return StatusResponse(status=Status.EXCEPTION,
details="Invalid temperature register selected")
class GetValueFromRegister(TransformationStrategy[int, int]):
"""Fetches the raw integer from the Modbus data list."""
def __init__(self, input_register: List[int]):
self.input_register = input_register
def transform(self, value: int) -> Response[int]:
raw_sensor_value = self.input_register[value]
return Response(status=Status.OK,
details=DEFAULT_SUCCESS_MESSAGE, value=raw_sensor_value)
class DetectSensorErrors(ValidationStrategy[int]):
"""Checks for hardware-specific error codes."""
NO_SENSOR = -32768
SENSOR_SHORT = 32767
def validate(self, value: int) -> StatusResponse:
if value == self.NO_SENSOR:
return StatusResponse(status=Status.EXCEPTION,
details="No sensor detected")
if value == self.SENSOR_SHORT:
return StatusResponse(status=Status.EXCEPTION,
details="Sensor short circuit")
return StatusResponse(status=Status.OK,
details=DEFAULT_SUCCESS_MESSAGE)
class RawToCelsius(TransformationStrategy[int, float]):
"""Transforms the raw integer to a float in degrees Celsius."""
def transform(self, value: int) -> Response[float]:
celsius = float(value) / 10.0
return Response(status=Status.OK,
details=DEFAULT_SUCCESS_MESSAGE, value=celsius)
class VentilationTemperature(RangeValue[float]):
"""
This value object encapsulates the full pipeline of reading and validating
temperature data from Modbus input registers, converting to Celsius, and
enforcing an allowed range.
"""
__slots__ = ("_getValueFromRegister",)
def __init__(self, input_register: Response[int], selected_register: int):
object.__setattr__(self,
"_getValueFromRegister",
GetValueFromRegister(input_register))
super().__init__(selected_register, -10.0, 40.0)
def get_strategies(self) -> List[PipeLineStrategy]:
return [TypeValidationStrategy(int),
AllowedInputRegister(),
self._getValueFromRegister,
DetectSensorErrors(),
RawToCelsius()] + super().get_strategies()
def main():
registers = [215, -32768, 32767, 402] # Example Modbus register values
print("=== Valid register 0 ===")
v = VentilationTemperature(registers, 0)
print(f"status={v.status}, details={v.details}, value={v.value}")
print("\n=== Invalid: No sensor detected (register 1) ===")
v = VentilationTemperature(registers, 1)
print(f"status={v.status}, details={v.details}")
print("\n=== Invalid: Sensor short circuit (register 2) ===")
v = VentilationTemperature(registers, 2)
print(f"status={v.status}, details={v.details}")
print("\n=== Out of range (register 3) ===")
v = VentilationTemperature(registers, 3)
print(f"status={v.status}, details={v.details}")
Output
=== Valid register 0 ===
status=Status.OK, details=validation successful, value=21.5
=== Invalid: No sensor detected (register 1) ===
status=Status.EXCEPTION, details=No sensor detected
=== Invalid: Sensor short circuit (register 2) ===
status=Status.EXCEPTION, details=Sensor short circuit
=== Out of range (register 3) ===
status=Status.EXCEPTION, details=Value must be less than or equal to 40.0, got 40.2
Examples Gallery
Explore all 31 runnable examples included with constrained_values.
- Source code: examples/ on GitHub
π§© Core Value Semantics
- Value Equality and Hashing Β· source
- Value Ordering (Same Class) Β· source
- Pass-Through ConstrainedValue Β· source
- Constrained Failure Β· source
- Truthiness and
.okΒ· source - Using
.unwrap()Β· source __str__and__format__Β· source- Hashing: Valid vs Invalid Values Β· source
- Chained Transforms Β· source
βοΈ Validation Strategies
π¨ Enums & Membership
- Enum with Class Β· source
- Enum with Members Sequence Β· source
- Enum with Plain Values Β· source
- Enum Config Errors (No Throw) Β· source
π Coercion & Sanitization
π¦ Strict & Ordering Behavior
- Strict Validated Value Β· source
- Sorting with Invalid Values Β· source
- Cross-Class Ordering TypeError Β· source
π§ Advanced Pipelines & Integrations
- Unknown Strategy Handler Β· source
- App Config Validation Β· source
- Money Amount Parsing Β· source
- UUID Parsing Β· source
- URL Value Β· source
- Email Value Β· source
- Date Parsing and Range Validation Β· source
- Record List Validation Β· source
- Pipeline Logging Strategy Β· source
- Dataclass Integration Β· source
API Documentation
1""" 2.. include:: ../docs/introduction.md 3.. include:: ../docs/examples.md 4""" 5from .constants import DEFAULT_SUCCESS_MESSAGE 6from .status import Status 7from .response import Response, T 8from .value import Value, ValidationStrategy, ConstrainedValue 9from .constrained_value_types import ( 10 EnumValue, 11 RangeValue, 12 StrictValue, 13) 14from .strategies import ( 15 TypeValidationStrategy, 16 RangeValidationStrategy, 17 EnumValidationStrategy, 18) 19 20__version__ = "0.1.7" 21 22__all__ = [ 23 "Value", 24 "ConstrainedValue", 25 "ValidationStrategy", 26 "TypeValidationStrategy", 27 "RangeValidationStrategy", 28 "EnumValidationStrategy", 29 "EnumValue", 30 "RangeValue", 31 "StrictValue", 32 "DEFAULT_SUCCESS_MESSAGE", 33 "Status", 34 "Response", 35]
35@dataclass(frozen=True, slots=True) 36class Value(Generic[T]): 37 """Immutable, typed value wrapper with same-class comparison semantics. 38 39 A lightweight wrapper around a value of type ``T``. Instances compare for 40 equality and ordering **only** against the same concrete subclass; cross-class 41 comparisons return ``NotImplemented`` (so Python can try reversed ops). 42 43 The dataclass is ``frozen=True`` for immutability and uses ``__slots__`` to 44 reduce memory footprint. 45 46 Type Variables: 47 T: The canonical type of the wrapped value. 48 49 Attributes: 50 _value (T): The wrapped value (read-only). 51 52 Example: 53 >>> a = Value(10) 54 >>> b = Value(10) 55 >>> c = Value(20) 56 >>> print("a == b:", a == b) 57 a == b: True 58 >>> print("a == c:", a == c) 59 a == c: False 60 >>> print("hash(a) == hash(b):", hash(a) == hash(b)) 61 hash(a) == hash(b): True 62 >>> class IntValue(Value[int]): pass 63 >>> class StrValue(Value[str]): pass 64 >>> print("IntValue(5) == StrValue('5'):", IntValue(5) == StrValue("5")) 65 IntValue(5) == StrValue('5'): False 66 """ 67 68 # Stored payload; immutable thanks to frozen dataclass 69 _value: T 70 71 def _class_is_same(self, other) -> bool: 72 """Return True if `other` is the same concrete subclass as `self`.""" 73 return other.__class__ is self.__class__ 74 75 def __repr__(self): 76 """Use !r to ensure the underlying value is shown with its repr() form 77 (developer-friendly and unambiguous, e.g., Value('foo') instead of Value(foo)). 78 """ 79 return f"{self.__class__.__name__}({self.value!r})" 80 81 @property 82 def value(self) -> T: 83 """Returns the stored value.""" 84 return self._value 85 86 def _compare(self, other: Value[T], comparison_func: Callable[[T, T], bool]) -> bool | NotImplementedType: 87 """Compare values using the provided function if classes match. 88 89 Args: 90 other (Value[T]): The other value instance to compare. 91 comparison_func (Callable[[T, T], bool]): The comparator used when classes match. 92 93 Returns: 94 bool | NotImplementedType: Result of the comparison or NotImplemented for cross-class. 95 """ 96 if self._class_is_same(other): 97 return comparison_func(self.value, other.value) 98 return NotImplemented 99 100 def __eq__(self, other): 101 return self._compare(other, lambda x, y: x == y) 102 103 def __lt__(self, other): 104 res = self._compare(other, lambda x, y: x < y) 105 return NotImplemented if res is NotImplemented else res 106 107 def __le__(self, other): 108 res = self._compare(other, lambda x, y: x <= y) 109 return NotImplemented if res is NotImplemented else res 110 111 def __gt__(self, other): 112 res = self._compare(other, lambda x, y: x > y) 113 return NotImplemented if res is NotImplemented else res 114 115 def __ge__(self, other): 116 res = self._compare(other, lambda x, y: x >= y) 117 return NotImplemented if res is NotImplemented else res 118 119 def __hash__(self): 120 return hash((self.__class__, self._value)) 121 122 def __str__(self) -> str: 123 return str(self.value) 124 125 def __format__(self, format_spec: str) -> str: 126 # Delegate formatting to the underlying value 127 return format(self.value, format_spec)
Immutable, typed value wrapper with same-class comparison semantics.
A lightweight wrapper around a value of type T. Instances compare for
equality and ordering only against the same concrete subclass; cross-class
comparisons return NotImplemented (so Python can try reversed ops).
The dataclass is frozen=True for immutability and uses __slots__ to
reduce memory footprint.
Type Variables:
T: The canonical type of the wrapped value.
Attributes:
- _value (T): The wrapped value (read-only).
Example:
>>> a = Value(10) >>> b = Value(10) >>> c = Value(20) >>> print("a == b:", a == b) a == b: True >>> print("a == c:", a == c) a == c: False >>> print("hash(a) == hash(b):", hash(a) == hash(b)) hash(a) == hash(b): True >>> class IntValue(Value[int]): pass >>> class StrValue(Value[str]): pass >>> print("IntValue(5) == StrValue('5'):", IntValue(5) == StrValue("5")) IntValue(5) == StrValue('5'): False
187class ConstrainedValue(Value[T], ABC): 188 """A value that is validated/transformed by a processing pipeline. 189 190 A :class:`ConstrainedValue` accepts raw input (``InT``) and runs it through a 191 configured sequence of :class:`PipeLineStrategy` steps (transformations and 192 validations) to produce a canonical value of type ``T`` and a validation 193 status. 194 195 Type Variables: 196 T: The canonical (final) type after pipeline processing. 197 InT: The raw input type supplied to the constructor. 198 199 Private Attributes: 200 _status (Status): Current status (OK or EXCEPTION). 201 _details (str): Human-readable details about the validation / transformation result. 202 203 Properties: 204 value (Optional[T]): The canonical value if valid, else ``None``. 205 ok (bool): True when status is OK. 206 207 Raises: 208 ValueError: When calling :meth:`unwrap` on an invalid instance. 209 Example: 210 >>> class Strip(TransformationStrategy[str, str]): 211 ... def transform(self, value: str) -> Response[str]: 212 ... return Response(Status.OK, "stripped", value.strip()) 213 >>> 214 >>> class NonEmpty(ValidationStrategy[str]): 215 ... def validate(self, value: str) -> StatusResponse: 216 ... return StatusResponse(Status.OK, "ok") 217 ... if value else StatusResponse(Status.EXCEPTION, "empty") 218 >>> 219 >>> class Name(ConstrainedValue[str]): 220 ... def get_strategies(self): return [Strip(), NonEmpty()] 221 >>> 222 >>> a = Name(" Alice ") 223 >>> b = Name("Alice") 224 >>> c = Name(" ") 225 >>> print(a.value, a.ok, a == b, bool(c)) 226 Alice True True False 227 >>> print(c) 228 <invalid Name: empty> 229 """ 230 def __repr__(self): 231 """Developer-friendly representation including value and status.""" 232 return f"{self.__class__.__name__}(_value={self._value!r}, status={self.status.name})" 233 234 __slots__ = ("_status", "_details") 235 236 def __init__(self, value_in: InT, success_details: str = DEFAULT_SUCCESS_MESSAGE): 237 result = self._run_pipeline(value_in, success_details) 238 super().__init__(result.value) 239 object.__setattr__(self, "_status", result.status) 240 object.__setattr__(self, "_details", result.details) 241 242 @classmethod 243 def _apply_strategy(cls, strategy: PipeLineStrategy, current_value: Any) -> Response[Any]: 244 """Run a single pipeline strategy and normalize the result. 245 246 Behavior: 247 - For transformations: return the strategy's Response. 248 - For validations: wrap the StatusResponse into a Response that carries the current value unchanged. 249 250 Args: 251 strategy (PipeLineStrategy): The pipeline strategy to apply. 252 current_value (Any): The current value being threaded through the pipeline. 253 254 Returns: 255 Response[Any]: A normalized response containing status, details, and value. 256 """ 257 if isinstance(strategy, TransformationStrategy): 258 return strategy.transform(current_value) 259 elif isinstance(strategy, ValidationStrategy): 260 # ValidationStrategy: keep the current value unchanged 261 sr = strategy.validate(current_value) 262 return Response(status=sr.status, details=sr.details, value=current_value) 263 return Response(status=Status.EXCEPTION, details="Missing strategy handler", value=None) 264 265 def _run_pipeline(self, value_in: InT, success_details:str)-> Response[T]: 266 """Thread the current value through the configured pipeline. 267 268 Transformation steps may change the value (or type), while validation steps only 269 inspect it. On the first ``Status.EXCEPTION`` the pipeline short-circuits and 270 returns that failure response; otherwise, it returns OK with the final canonical value. 271 272 Args: 273 value_in (InT): The raw input value to start the pipeline. 274 success_details (str): Details message to use when the pipeline completes with OK. 275 276 Returns: 277 Response[T]: The terminal response (status, details, and canonical value). 278 """ 279 current_value = value_in # Start with the initial value 280 281 for strategy in self.get_strategies(): 282 resp = self._apply_strategy(strategy, current_value) 283 if resp.status == Status.EXCEPTION: 284 return Response(status=Status.EXCEPTION, details=resp.details, value=None) 285 # OK β thread the (possibly transformed) value 286 current_value = resp.value 287 288 return Response(status=Status.OK, details=success_details, value=current_value) 289 290 @abstractmethod 291 def get_strategies(self) -> List[PipeLineStrategy]: 292 """Return the ordered list of strategies for this pipeline. 293 294 Returns: 295 List[PipeLineStrategy]: Transformation and validation steps in the order to apply. 296 """ 297 ... 298 @property 299 def status(self) -> Status: 300 """Current status (OK or EXCEPTION).""" 301 return self._status 302 303 @property 304 def details(self) -> str: 305 """Human-readable details about the validation / transformation result.""" 306 return self._details 307 308 @property 309 def value(self) -> Optional[T]: 310 """Canonical value if valid; otherwise ``None`` when status is EXCEPTION.""" 311 if self._status == Status.EXCEPTION: 312 return None 313 return self._value 314 315 def _same_status(self, other): 316 """Return True if both instances share the same Status.""" 317 return self.status == other.status 318 319 def __eq__(self, other): 320 if not self._class_is_same(other): 321 return False 322 if self.status != Status.OK or other.status != Status.OK: 323 return False 324 return super().__eq__(other) 325 326 def _is_comparing(self, other: ConstrainedValue[T], 327 func: Callable[[Value[T]], bool | NotImplementedType]): 328 """Perform a comparison with another ConstrainedValue instance. 329 330 Internal helper for ordering comparisons: 331 - Ensures the same concrete subclass. 332 - Ensures both operands are valid (Status.OK). 333 - Delegates the comparison to the base Value comparator. 334 335 Args: 336 other (ConstrainedValue[T]): The other instance to compare against. 337 func (Callable[[Value[T]], bool | NotImplementedType]): The comparison 338 function to delegate to (e.g., ``super().__lt__``). 339 340 Returns: 341 bool | NotImplementedType: The result of the comparison if valid, or 342 ``NotImplemented`` if the instances are not comparable. 343 344 Raises: 345 ValueError: If either operand has a non-OK validation status. 346 """ 347 if not self._class_is_same(other): 348 return NotImplemented 349 if self.status != Status.OK or other.status != Status.OK: 350 raise ValueError(f"{self.__class__.__name__}: cannot compare invalid values") 351 return func(other) 352 353 def __lt__(self, other): 354 return self._is_comparing(other, super().__lt__) 355 356 def __le__(self, other): 357 return self._is_comparing(other, super().__le__) 358 359 def __gt__(self, other): 360 return self._is_comparing(other, super().__gt__) 361 362 def __ge__(self, other): 363 return self._is_comparing(other, super().__ge__) 364 365 def __hash__(self): 366 if self.status == Status.OK: 367 # Match value-based equality for valid instances 368 return hash((self.__class__, self._value)) 369 # For invalid instances: still hashable, but distinct from any valid instance 370 # Donβt include .details (too volatile); status is enough. 371 return hash((self.__class__, self.status)) 372 373 def __bool__(self) -> bool: 374 return self.status == Status.OK 375 376 def __str__(self) -> str: 377 # Print the canonical value when valid; show a concise marker when invalid 378 return str(self._value) if self.status == Status.OK else f"<invalid {self.__class__.__name__}: {self.details}>" 379 380 # Ensures invalid values format to the same marker as __str__ (not "None"). 381 # This keeps f-strings readable even when the instance is invalid. 382 def __format__(self, format_spec: str) -> str: 383 if self.status == Status.OK: 384 return format(self._value, format_spec) 385 return str(self) 386 387 def unwrap(self) -> T: 388 """Return the validated value or raise if invalid (ergonomic for callers).""" 389 if self.status != Status.OK: 390 raise ValueError(f"{self.__class__.__name__} invalid: {self.details}") 391 return self._value 392 393 @property 394 def ok(self) -> bool: 395 """Convenience alias for status == Status.OK.""" 396 return self.status == Status.OK
A value that is validated/transformed by a processing pipeline.
A ConstrainedValue accepts raw input (InT) and runs it through a
configured sequence of PipeLineStrategy steps (transformations and
validations) to produce a canonical value of type T and a validation
status.
Type Variables:
T: The canonical (final) type after pipeline processing. InT: The raw input type supplied to the constructor.
Private Attributes:
_status (Status): Current status (OK or EXCEPTION). _details (str): Human-readable details about the validation / transformation result.
Properties:
value (Optional[T]): The canonical value if valid, else
None. ok (bool): True when status is OK.
Raises:
- ValueError: When calling
unwrap()on an invalid instance.
Example:
>>> class Strip(TransformationStrategy[str, str]): ... def transform(self, value: str) -> Response[str]: ... return Response(Status.OK, "stripped", value.strip()) >>> >>> class NonEmpty(ValidationStrategy[str]): ... def validate(self, value: str) -> StatusResponse: ... return StatusResponse(Status.OK, "ok") ... if value else StatusResponse(Status.EXCEPTION, "empty") >>> >>> class Name(ConstrainedValue[str]): ... def get_strategies(self): return [Strip(), NonEmpty()] >>> >>> a = Name(" Alice ") >>> b = Name("Alice") >>> c = Name(" ") >>> print(a.value, a.ok, a == b, bool(c)) Alice True True False >>> print(c) <invalid Name: empty>
290 @abstractmethod 291 def get_strategies(self) -> List[PipeLineStrategy]: 292 """Return the ordered list of strategies for this pipeline. 293 294 Returns: 295 List[PipeLineStrategy]: Transformation and validation steps in the order to apply. 296 """ 297 ...
Return the ordered list of strategies for this pipeline.
Returns:
List[PipeLineStrategy]: Transformation and validation steps in the order to apply.
298 @property 299 def status(self) -> Status: 300 """Current status (OK or EXCEPTION).""" 301 return self._status
Current status (OK or EXCEPTION).
303 @property 304 def details(self) -> str: 305 """Human-readable details about the validation / transformation result.""" 306 return self._details
Human-readable details about the validation / transformation result.
308 @property 309 def value(self) -> Optional[T]: 310 """Canonical value if valid; otherwise ``None`` when status is EXCEPTION.""" 311 if self._status == Status.EXCEPTION: 312 return None 313 return self._value
Canonical value if valid; otherwise None when status is EXCEPTION.
387 def unwrap(self) -> T: 388 """Return the validated value or raise if invalid (ergonomic for callers).""" 389 if self.status != Status.OK: 390 raise ValueError(f"{self.__class__.__name__} invalid: {self.details}") 391 return self._value
Return the validated value or raise if invalid (ergonomic for callers).
141class ValidationStrategy(Generic[MidT], PipeLineStrategy): 142 """Abstract base class for validation strategies. 143 144 Subclasses should implement :meth:`validate` to examine the provided value 145 and return a :class:`~constrained_values.response.StatusResponse` describing 146 whether the value is valid. 147 148 Type variables: 149 MidT: The type of value accepted by this validator. 150 """ 151 152 @abstractmethod 153 def validate(self, value: MidT) -> StatusResponse: # pragma: no cover 154 """Validate ``value`` and return a StatusResponse. 155 156 Args: 157 value (MidT): Value to validate. 158 159 Returns: 160 StatusResponse: Validation outcome (status and details). 161 """ 162 pass
Abstract base class for validation strategies.
Subclasses should implement validate() to examine the provided value
and return a ~constrained_values.response.StatusResponse describing
whether the value is valid.
Type variables:
MidT: The type of value accepted by this validator.
152 @abstractmethod 153 def validate(self, value: MidT) -> StatusResponse: # pragma: no cover 154 """Validate ``value`` and return a StatusResponse. 155 156 Args: 157 value (MidT): Value to validate. 158 159 Returns: 160 StatusResponse: Validation outcome (status and details). 161 """ 162 pass
Validate value and return a StatusResponse.
Arguments:
- value (MidT): Value to validate.
Returns:
StatusResponse: Validation outcome (status and details).
35class TypeValidationStrategy(ValidationStrategy[Any]): 36 """Validation strategy to ensure the runtime "type" of a value is one of the allowed types. 37 38 Args: 39 valid_types (type or Sequence[type]): Allowed types for validation. 40 """ 41 42 def __init__(self, valid_types: Sequence[type] | type): 43 self.valid_types: Tuple[type, ...] = get_types(valid_types) 44 45 def validate(self, value: Any) -> StatusResponse: 46 """Validate that the value is of one of the allowed types. 47 48 Args: 49 value (Any): The value to validate. 50 51 Returns: 52 StatusResponse: Validation result with status and details. 53 """ 54 if type(value) not in self.valid_types: 55 types_str = ", ".join(f"'{t.__name__}'" for t in self.valid_types) 56 return StatusResponse( 57 status=Status.EXCEPTION, 58 details=f"Value must be one of {types_str}, got '{type(value).__name__}'" 59 ) 60 return StatusResponse(status=Status.OK, details=DEFAULT_SUCCESS_MESSAGE)
Validation strategy to ensure the runtime "type" of a value is one of the allowed types.
Arguments:
- valid_types (type or Sequence[type]): Allowed types for validation.
45 def validate(self, value: Any) -> StatusResponse: 46 """Validate that the value is of one of the allowed types. 47 48 Args: 49 value (Any): The value to validate. 50 51 Returns: 52 StatusResponse: Validation result with status and details. 53 """ 54 if type(value) not in self.valid_types: 55 types_str = ", ".join(f"'{t.__name__}'" for t in self.valid_types) 56 return StatusResponse( 57 status=Status.EXCEPTION, 58 details=f"Value must be one of {types_str}, got '{type(value).__name__}'" 59 ) 60 return StatusResponse(status=Status.OK, details=DEFAULT_SUCCESS_MESSAGE)
Validate that the value is of one of the allowed types.
Arguments:
- value (Any): The value to validate.
Returns:
StatusResponse: Validation result with status and details.
97class RangeValidationStrategy(ValidationStrategy[Any]): 98 """Validation strategy to ensure a value is within a specified range [low_value, high_value]. 99 100 Args: 101 low_value (Any): The lower bound (inclusive). 102 high_value (Any): The upper bound (inclusive). 103 """ 104 105 def __init__(self, low_value: Any, high_value: Any): 106 self.low_value = low_value 107 self.high_value = high_value 108 109 def validate(self, value: Any) -> StatusResponse: 110 """Validate that the value is within the specified range. 111 112 Args: 113 value (Any): The value to validate. 114 115 Returns: 116 StatusResponse: Validation result with status and details. 117 """ 118 if value < self.low_value: 119 return StatusResponse( 120 status=Status.EXCEPTION, 121 details=f"Value must be greater than or equal to {self.low_value}, got {value}" 122 ) 123 if value > self.high_value: 124 return StatusResponse( 125 status=Status.EXCEPTION, 126 details=f"Value must be less than or equal to {self.high_value}, got {value}" 127 ) 128 return StatusResponse(status=Status.OK, details=DEFAULT_SUCCESS_MESSAGE)
Validation strategy to ensure a value is within a specified range [low_value, high_value].
Arguments:
- low_value (Any): The lower bound (inclusive).
- high_value (Any): The upper bound (inclusive).
109 def validate(self, value: Any) -> StatusResponse: 110 """Validate that the value is within the specified range. 111 112 Args: 113 value (Any): The value to validate. 114 115 Returns: 116 StatusResponse: Validation result with status and details. 117 """ 118 if value < self.low_value: 119 return StatusResponse( 120 status=Status.EXCEPTION, 121 details=f"Value must be greater than or equal to {self.low_value}, got {value}" 122 ) 123 if value > self.high_value: 124 return StatusResponse( 125 status=Status.EXCEPTION, 126 details=f"Value must be less than or equal to {self.high_value}, got {value}" 127 ) 128 return StatusResponse(status=Status.OK, details=DEFAULT_SUCCESS_MESSAGE)
Validate that the value is within the specified range.
Arguments:
- value (Any): The value to validate.
Returns:
StatusResponse: Validation result with status and details.
131class EnumValidationStrategy(ValidationStrategy[Any]): 132 """Validation strategy to ensure a value is one of a provided collection. 133 134 Args: 135 valid_values (Sequence[Any]): The collection of valid values. 136 """ 137 138 def __init__(self, valid_values: Sequence[Any]): 139 self.valid_values = valid_values 140 141 def validate(self, value: Any) -> StatusResponse: 142 """Validate that the value is in the collection of valid values. 143 144 Args: 145 value (Any): The value to validate. 146 147 Returns: 148 StatusResponse: Validation result with status and details. 149 """ 150 if value not in self.valid_values: 151 return StatusResponse( 152 status=Status.EXCEPTION, 153 details=f"Value must be one of {self.valid_values}, got {value}" 154 ) 155 return StatusResponse(status=Status.OK, details=DEFAULT_SUCCESS_MESSAGE)
Validation strategy to ensure a value is one of a provided collection.
Arguments:
- valid_values (Sequence[Any]): The collection of valid values.
141 def validate(self, value: Any) -> StatusResponse: 142 """Validate that the value is in the collection of valid values. 143 144 Args: 145 value (Any): The value to validate. 146 147 Returns: 148 StatusResponse: Validation result with status and details. 149 """ 150 if value not in self.valid_values: 151 return StatusResponse( 152 status=Status.EXCEPTION, 153 details=f"Value must be one of {self.valid_values}, got {value}" 154 ) 155 return StatusResponse(status=Status.OK, details=DEFAULT_SUCCESS_MESSAGE)
Validate that the value is in the collection of valid values.
Arguments:
- value (Any): The value to validate.
Returns:
StatusResponse: Validation result with status and details.
70class EnumValue(ConstrainedValue[T]): 71 """Validates enum-like values using a pipeline of strategies. 72 73 Pipeline: 74 1. CoerceEnumMemberToValue: Converts Enum members to their values (if needed). 75 2. TypeValidationStrategy: Ensures value is of allowed types. 76 3. EnumValidationStrategy: Checks membership in allowed values. 77 78 If configuration is invalid (e.g., empty enum/sequence), FailValidationStrategy is used to surface an error. 79 80 Args: 81 value (object): The value to validate. 82 valid_values (Sequence[T] | Type[Enum]): Allowed values or Enum type. 83 success_details (str, optional): Details for successful validation. Defaults to DEFAULT_SUCCESS_MESSAGE. 84 85 Example: 86 >>> EnumValue('A', ['A', 'B', 'C']) 87 """ 88 __slots__ = ("_strategies",) 89 90 @classmethod 91 def _all_enum_members(cls, seq: Sequence[Any]) -> bool: 92 return len(seq) > 0 and all(isinstance(x, Enum) for x in seq) 93 94 @classmethod 95 def _normalize_allowed(cls, valid_values: Sequence[Any] | Type[Enum]) -> tuple[list[Any], bool, str | None]: 96 """Normalize 'valid_values' into: 97 - allowed_values: list of canonical values to check membership against 98 - needs_coercion: whether to add the enumβvalue coercion step 99 - error_details: None if OK; otherwise a message explaining an error 100 """ 101 seq = list(valid_values) 102 103 # Enum class 104 if isinstance(valid_values, type) and issubclass(valid_values, Enum): 105 if not seq: 106 return [], False, "Enum has no members." 107 return [m.value for m in seq], True, None 108 109 # Sequence 110 if not seq: 111 return [], False, "Must be a non-empty sequence." 112 113 # Sequence of Enum members 114 if EnumValue._all_enum_members(seq): 115 return [m.value for m in seq], True, None 116 117 # Plain values 118 return seq, False, None 119 120 def get_strategies(self) -> List[PipeLineStrategy]: 121 return self._strategies 122 123 def __init__( 124 self, 125 value: object, 126 valid_values: Sequence[T] | Type[Enum], 127 success_details: str = DEFAULT_SUCCESS_MESSAGE, 128 ): 129 allowed, needs_coercion, err = EnumValue._normalize_allowed(valid_values) 130 131 strategies: List[PipeLineStrategy] = [] 132 if err is None: 133 if needs_coercion: 134 strategies.append(CoerceEnumMemberToValue()) 135 strategies += [ 136 TypeValidationStrategy(types_of_values(allowed)), 137 EnumValidationStrategy(tuple(allowed)), 138 ] 139 else: 140 # Config problem β report as EXCEPTION through the pipeline (no throws) 141 strategies.append(FailValidationStrategy(err)) 142 143 object.__setattr__(self, "_strategies", strategies) 144 super().__init__(value, success_details)
Validates enum-like values using a pipeline of strategies.
Pipeline:
- CoerceEnumMemberToValue: Converts Enum members to their values (if needed).
- TypeValidationStrategy: Ensures value is of allowed types.
- EnumValidationStrategy: Checks membership in allowed values.
If configuration is invalid (e.g., empty enum/sequence), FailValidationStrategy is used to surface an error.
Arguments:
- value (object): The value to validate.
- valid_values (Sequence[T] | Type[Enum]): Allowed values or Enum type.
- success_details (str, optional): Details for successful validation. Defaults to DEFAULT_SUCCESS_MESSAGE.
Example:
>>> EnumValue('A', ['A', 'B', 'C'])
123 def __init__( 124 self, 125 value: object, 126 valid_values: Sequence[T] | Type[Enum], 127 success_details: str = DEFAULT_SUCCESS_MESSAGE, 128 ): 129 allowed, needs_coercion, err = EnumValue._normalize_allowed(valid_values) 130 131 strategies: List[PipeLineStrategy] = [] 132 if err is None: 133 if needs_coercion: 134 strategies.append(CoerceEnumMemberToValue()) 135 strategies += [ 136 TypeValidationStrategy(types_of_values(allowed)), 137 EnumValidationStrategy(tuple(allowed)), 138 ] 139 else: 140 # Config problem β report as EXCEPTION through the pipeline (no throws) 141 strategies.append(FailValidationStrategy(err)) 142 143 object.__setattr__(self, "_strategies", strategies) 144 super().__init__(value, success_details)
147class RangeValue(ConstrainedValue[T]): 148 """Constrained numeric value bounded between low_value and high_value (inclusive). 149 150 Validation/transform pipeline: 151 1. Type strategies: 152 - SameTypeValidationStrategy: Ensures bounds are of the same type. 153 - TypeValidationStrategy: Infers acceptable input types from bounds. 154 - CoerceToType: Coerces candidate to type of low_value. 155 2. Custom strategies: Hook for subclasses to inject additional logic. 156 3. Range strategies: 157 - RangeValidationStrategy: Enforces low_value <= value <= high_value. 158 159 Args: 160 value (Any): The value to validate. 161 low_value (Any): Lower bound (inclusive). 162 high_value (Any): Upper bound (inclusive). 163 success_details (str, optional): Details for successful validation. Defaults to DEFAULT_SUCCESS_MESSAGE. 164 165 Example: 166 >>> RangeValue(5, 1, 10) 167 168 Notes: 169 - Canonical values are always coerced to the type of low_value. 170 - If bounds are floats, both int and float inputs are accepted and coerced to float. 171 - If bounds are Decimals, both int and Decimal inputs are accepted and coerced to Decimal. 172 """ 173 __slots__ = ("_type_strategies", "_range_strategies") 174 175 @classmethod 176 def infer_valid_types_from_value(cls, value) -> Tuple[Type, ...]: 177 t = type(value) 178 if t is int: 179 return (int,) 180 if t is float: 181 return int, float 182 if t is Decimal: 183 return int, Decimal 184 if t is Fraction: 185 return int, Fraction 186 # default: exact type only 187 return (t,) 188 189 def get_strategies(self) -> List[PipeLineStrategy]: 190 return self._type_strategies + self.get_custom_strategies() + self._range_strategies 191 192 def get_custom_strategies(self) -> list[PipeLineStrategy]: 193 return [] 194 195 def __init__(self, value, low_value, high_value, success_details: str = DEFAULT_SUCCESS_MESSAGE): 196 # Initialize the strategies for this subclass 197 object.__setattr__(self, "_type_strategies", [ 198 SameTypeValidationStrategy(low_value, high_value), 199 TypeValidationStrategy(RangeValue.infer_valid_types_from_value(low_value)), 200 CoerceToType(type(low_value)) 201 ]) 202 object.__setattr__(self, "_range_strategies", [ 203 RangeValidationStrategy(low_value, high_value) 204 ]) 205 super().__init__(value, success_details)
Constrained numeric value bounded between low_value and high_value (inclusive).
Validation/transform pipeline: 1. Type strategies: - SameTypeValidationStrategy: Ensures bounds are of the same type. - TypeValidationStrategy: Infers acceptable input types from bounds. - CoerceToType: Coerces candidate to type of low_value. 2. Custom strategies: Hook for subclasses to inject additional logic. 3. Range strategies: - RangeValidationStrategy: Enforces low_value <= value <= high_value.
Arguments:
- value (Any): The value to validate.
- low_value (Any): Lower bound (inclusive).
- high_value (Any): Upper bound (inclusive).
- success_details (str, optional): Details for successful validation. Defaults to DEFAULT_SUCCESS_MESSAGE.
Example:
>>> RangeValue(5, 1, 10)
Notes:
- Canonical values are always coerced to the type of low_value.
- If bounds are floats, both int and float inputs are accepted and coerced to float.
- If bounds are Decimals, both int and Decimal inputs are accepted and coerced to Decimal.
195 def __init__(self, value, low_value, high_value, success_details: str = DEFAULT_SUCCESS_MESSAGE): 196 # Initialize the strategies for this subclass 197 object.__setattr__(self, "_type_strategies", [ 198 SameTypeValidationStrategy(low_value, high_value), 199 TypeValidationStrategy(RangeValue.infer_valid_types_from_value(low_value)), 200 CoerceToType(type(low_value)) 201 ]) 202 object.__setattr__(self, "_range_strategies", [ 203 RangeValidationStrategy(low_value, high_value) 204 ]) 205 super().__init__(value, success_details)
175 @classmethod 176 def infer_valid_types_from_value(cls, value) -> Tuple[Type, ...]: 177 t = type(value) 178 if t is int: 179 return (int,) 180 if t is float: 181 return int, float 182 if t is Decimal: 183 return int, Decimal 184 if t is Fraction: 185 return int, Fraction 186 # default: exact type only 187 return (t,)
189 def get_strategies(self) -> List[PipeLineStrategy]: 190 return self._type_strategies + self.get_custom_strategies() + self._range_strategies
Return the ordered list of strategies for this pipeline.
Returns:
List[PipeLineStrategy]: Transformation and validation steps in the order to apply.
209class StrictValue(ConstrainedValue[T], ABC): 210 """Stricter version of ConstrainedValue that raises an exception immediately if validation fails. 211 212 Args: 213 value (T, optional): The value to validate. Defaults to None. 214 success_details (str, optional): Details for successful validation. Defaults to DEFAULT_SUCCESS_MESSAGE. 215 216 Raises: 217 ValueError: If validation fails. 218 """ 219 def __init__(self, value: T = None, success_details: str = DEFAULT_SUCCESS_MESSAGE): 220 super().__init__(value, success_details) 221 if self.status == Status.EXCEPTION: 222 raise ValueError(f"Failed Constraints for value - '{value}': {self.details}")
Stricter version of ConstrainedValue that raises an exception immediately if validation fails.
Arguments:
- value (T, optional): The value to validate. Defaults to None.
- success_details (str, optional): Details for successful validation. Defaults to DEFAULT_SUCCESS_MESSAGE.
Raises:
- ValueError: If validation fails.
Inherited Members
Enum to represent the status of a process.
Inherited Members
- enum.Enum
- name
- value
22@dataclass(frozen=True) 23class Response(StatusResponse[T]): 24 """Data class to encapsulate the result of a process. 25 Attributes: 26 value: The value after a process is completed. 27 """ 28 29 value: Optional[T]
Data class to encapsulate the result of a process.
Attributes:
- value: The value after a process is completed.
Inherited Members
- constrained_values.response.StatusResponse
- status
- details