examples.30_pipeline_logging

  1"""Demonstrates how to record detailed step-by-step logs for every
  2transformation in a :class:`ConstrainedValue` pipeline.
  3
  4This example shows:
  5  * How each transformation step records both input and output values.
  6  * How to store these logs in a per-strategy list.
  7  * How exceptions are caught and logged without aborting the pipeline.
  8
  9Run directly:
 10
 11    python examples/30_pipeline_logging.py
 12
 13Expected output:
 14
 15    ('strip', 'OK', '  hi  ', 'hi')
 16    ('upper', 'OK', 'hi', 'HI')
 17    ('suffix', 'OK', 'HI', 'HI_X')
 18    final: OK HI_X
 19"""
 20
 21import sys
 22import pathlib
 23from typing import Any, List
 24
 25# ---------------------------------------------------------------------------
 26# Make repo root importable when running this file directly
 27# ---------------------------------------------------------------------------
 28sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
 29
 30from constrained_values import Response, Status
 31from constrained_values.value import TransformationStrategy, ConstrainedValue
 32
 33
 34class LogStep(TransformationStrategy[Any, Any]):
 35    """A transformation strategy that logs every step of the pipeline.
 36
 37    Attributes:
 38        tag (str): Label identifying this step (e.g., "strip" or "upper").
 39        fn (Callable[[Any], Any]): Transformation function applied to the value.
 40        logs (List[Tuple[str, str, Any, Any]]): Recorded results per invocation.
 41    """
 42
 43    def __init__(self, tag: str, fn):
 44        """Initialize the logging strategy.
 45
 46        Args:
 47            tag: Human-readable identifier for the step.
 48            fn: Callable performing the transformation, e.g. `lambda s: s.upper()`.
 49        """
 50        self.tag = tag
 51        self.fn = fn
 52        self.logs: List[tuple] = []
 53
 54    def transform(self, value: Any) -> Response[Any]:
 55        """Apply the transformation and record the outcome.
 56
 57        Args:
 58            value: The input value passed down the pipeline.
 59
 60        Returns:
 61            Response[Any]: Successful or failed transformation result.
 62        """
 63        try:
 64            new_value = self.fn(value)
 65            self.logs.append((self.tag, "OK", value, new_value))
 66            return Response(Status.OK, self.tag, new_value)
 67        except Exception as e:
 68            self.logs.append((self.tag, "EX", value, str(e)))
 69            return Response(Status.EXCEPTION, f"{self.tag}: {e}", None)
 70
 71
 72class LoggedValue(ConstrainedValue[Any]):
 73    """A `ConstrainedValue` subclass that uses logging strategies.
 74
 75    The pipeline for this example:
 76        1. `strip` — remove leading/trailing spaces.
 77        2. `upper` — convert to uppercase.
 78        3. `suffix` — append `"_X"`.
 79    """
 80
 81    __slots__ = ("_strategies",)
 82
 83    def __init__(self, value: Any):
 84        """Initialize the pipeline with predefined logging steps."""
 85        steps: List[LogStep] = [
 86            LogStep("strip", lambda s: s.strip()),
 87            LogStep("upper", lambda s: s.upper()),
 88            LogStep("suffix", lambda s: s + "_X"),
 89        ]
 90        object.__setattr__(self, "_strategies", steps)
 91        super().__init__(value)
 92
 93    def get_strategies(self) -> List[LogStep]:
 94        """Return the pipeline of logging strategies."""
 95        return self._strategies
 96
 97
 98def main() -> None:
 99    """Run the pipeline logging demonstration.
100
101    Creates a `LoggedValue` from `"  hi  "` and prints all recorded logs.
102
103    Each pipeline step stores a tuple of:
104        `(step_name, status, input_value, output_value)`
105
106    Prints:
107        * Each logged tuple from every strategy.
108        * `"final: OK HI_X"` — the overall result.
109    """
110    x = LoggedValue("  hi  ")
111    for s in x.get_strategies():
112        for entry in s.logs:
113            print(entry)
114    print("final:", x.status.name, x.value)
115
116
117if __name__ == "__main__":
118    main()