Skip to content

Technical Reference

pipeline.core.pipe.pipe

Pipe

Bases: Generic[V, T]

A Pipe processes a single value through validation, matching, and transformation steps.

The execution flow is strict: 1. Optional Check: If the pipe is optional and the value is falsy, it returns early. 2. Type Validation: Checks if the value matches the expected type (via Condition.ValueType). 3. Setup: Executes setup (transform) handlers (e.g. Strip) to modify the value. Only the value type is validated prior to this stage. 4. Conditions: Runs a set of condition handlers. If any fail, errors are collected, and processing may stop if BREAK_PIPE_LOOP_ON_ERROR flag is set. 5. Matches: Only if no condition errors occurred, match handlers are executed. These are typically regex-based checks. 6. Transform: Only if no match errors occurred, transform handlers are executed to modify the value.

Attributes:

Name Type Description
Condition Type[Condition]

The condition handler class registry.
Match Type[Match]

The match handler class registry.
Transform Type[Transform]

The transform handler class registry.
Source code in pipeline/core/pipe/pipe.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
class Pipe(Generic[V, T]):
    """
    A Pipe processes a single value through validation, matching, and transformation steps.

    The execution flow is strict:
    1. Optional Check: If the pipe is optional and the value is falsy, it returns early.
    2. Type Validation: Checks if the value matches the expected type (via `Condition.ValueType`).
    3. Setup: Executes setup (transform) handlers (e.g. Strip) to modify the value. Only the
       value type is validated prior to this stage.
    4. Conditions: Runs a set of condition handlers. If any fail, errors are collected, and processing
       may stop if `BREAK_PIPE_LOOP_ON_ERROR` flag is set.
    5. Matches: Only if no condition errors occurred, match handlers are executed. These are typically
       regex-based checks.
    6. Transform: Only if no match errors occurred, transform handlers are executed to modify the value.

    Attributes:
        Condition (Type[Condition]): The condition handler class registry.
        Match (Type[Match]): The match handler class registry.
        Transform (Type[Transform]): The transform handler class registry.
    """
    Condition: ClassVar[Type[Condition]] = Condition
    Match: ClassVar[Type[Match]] = Match
    Transform: ClassVar[Type[Transform]] = Transform

    def __init__(
        self,
        value: V,
        type: T,
        setup: Optional[PipeTransform] = None,
        conditions: Optional[PipeConditions] = None,
        matches: Optional[PipeMatches] = None,
        transform: Optional[PipeTransform] = None,
        optional: Optional[bool] = None,
        context: Optional[PipeContext] = None,
        metadata: Optional[PipeMetadata] = None
    ) -> None:
        """
        Initializes the Pipe with a value, type, and processing configurations.

        Args:
            value (V): The value to process.
            type (T): The expected type of the value (e.g., `str`, `int`).
            setup (Optional[PipeTransform]): A dictionary of transform handlers and their arguments.
                Used for data setup (e.g. Strip). Use with caution. Setup runs after type validation, but
                before conditions, matches, and transform handlers. Only the value type is
                validated at the time of setup execution.
            conditions (Optional[PipeConditions]): A dictionary of condition handlers and their arguments.
                Used for logical validation (e.g., `MinLength`, `Equal`).
            matches (Optional[PipeMatches]): A dictionary of match handlers and their arguments.
                Used for pattern matching (e.g., `Email`, `Regex`).
            transform (Optional[PipeTransform]): A dictionary of transform handlers and their arguments.
                Used for data modification (e.g., `Strip`, `Capitalize`).
            optional (Optional[bool]): If True, the pipe is skipped if the value is falsy.
            context (Optional[PipeContext]): Additional context for the handlers, typically the
                entire data dictionary being processed.
            metadata (Optional[PipeMetadata]): Metadata about the pipe execution.
        """
        self.value: V = value

        self.type: T = type

        self.setup: Optional[PipeTransform] = setup

        self.conditions: Optional[PipeConditions] = conditions
        self.matches: Optional[PipeMatches] = matches
        self.transform: Optional[PipeTransform] = transform

        self.optional: Optional[bool] = optional

        self.context: Optional[PipeContext] = context
        self.metadata: Optional[PipeMetadata] = metadata

        self._condition_errors: ConditionErrors = []
        self._match_errors: ConditionErrors = []

    def run(self) -> PipeResult:
        """
        Executes the pipe processing logic.

        The method orchestrates the execution flow by delegating to specialized helper methods:
        1. Checks if validation should be skipped (optional pipe with falsy value)
        2. Validates the value type
        3. Processes setup handlers (if any)
        4. Processes condition handlers
        5. Processes match handlers (only if no condition errors)
        6. Processes transform handlers (only if no condition or match errors)

        Note that transformations are ONLY applied if all validations (Conditions and Matches) pass.
        This provides a safe way to transform data, ensuring it is valid first.

        Returns:
            PipeResult[V]: The result containing the processed value (or original value if errors occurred)
            and lists of any condition or match errors.
        """
        if self._should_skip_validation():
            return self._construct_result()

        if not self._is_value_type_correct():
            return self._construct_result()

        self._process_setup()

        self._process_conditions()
        self._process_matches()
        self._process_transform()

        return self._construct_result()

    def _construct_result(self) -> PipeResult:
        """
        Constructs and returns the final PipeResult.

        If the pipe is marked as optional and the value is falsy, 
        the result value is set to None.

        Returns:
            PipeResult: A result object containing the current value (or None 
            if optional and empty) and any accumulated condition or match errors.
        """
        value: V | None = None if self.optional and not bool(
            self.value
        ) else self.value

        return PipeResult(
            value=value,
            condition_errors=self._condition_errors,
            match_errors=self._match_errors
        )

    def _should_skip_validation(self) -> bool:
        """
        Determines whether validation should be skipped for this pipe.

        Validation is skipped only if the pipe is marked as optional and the value is falsy
        (e.g., None, empty string, 0, False).

        Returns:
            bool: True if validation should be skipped, False otherwise.
        """
        if not self.optional:
            return False

        return not bool(self.value)

    def _is_value_type_correct(self) -> bool:
        """
        Validates that the value matches the expected type.

        Uses the `Condition.ValueType` handler to check type correctness. If the type
        is incorrect, an error is appended to the condition errors list.

        Returns:
            bool: True if the value type is correct, False otherwise.
        """
        if (error := self.Condition.ValueType(self.value, self.type).handle()):
            self._condition_errors.append(error)

        return not error

    def _process_setup(self) -> None:
        """
        Processes setup transform handlers to prepare the value for validation.

        Setup handlers are executed after type validation but before conditions, matches,
        and transform handlers. They are typically used for data normalization (e.g., Strip).
        Each handler modifies the value in place.

        Note:
            Setup handlers run even if subsequent validations might fail. Use with caution.
        """
        if not self.setup:
            return

        for handler, argument in self.setup.items():
            self.value = handler(
                value=self.value, argument=argument, context=self.context
            ).handle()

    def _process_conditions(self) -> None:
        """
        Processes condition handlers to validate the value.

        Iterates through all condition handlers and executes them. If a handler returns an error,
        it is appended to the condition errors list. If a handler has the `BREAK_PIPE_LOOP_ON_ERROR`
        flag set, the loop terminates immediately upon encountering an error.

        Note:
            Condition errors prevent match and transform handlers from executing.
        """
        if not self.conditions:
            return

        for handler, argument in self.conditions.items():
            handler = handler(
                value=self.value, argument=argument, context=self.context
            )

            if (error := handler.handle()):
                self._condition_errors.append(error)

                if ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR in handler.FLAGS:
                    break

    def _process_matches(self) -> None:
        """
        Processes match handlers to perform pattern-based validation.

        Match handlers are only executed if no condition errors occurred. Typically used for
        regex-based validation (e.g., Email, URL patterns). The loop terminates immediately
        upon the first match error.

        Note:
            This method is skipped if any condition errors exist. Match errors prevent
            transform handlers from executing.
        """
        if not self.matches or len(self._condition_errors) != 0:
            return

        for handler, argument in self.matches.items():
            handler = handler(
                value=self.value, argument=argument, context=self.context
            )

            if (error := handler.handle()):
                self._match_errors.append(error)

                break

    def _process_transform(self) -> None:
        """
        Processes transform handlers to modify the value.

        Transform handlers are only executed if no condition or match errors occurred.
        This ensures that transformations are only applied to valid data. Each handler
        modifies the value in place.

        Note:
            This method is skipped if any condition or match errors exist, ensuring
            transformations are only applied to validated data.
        """
        if not self.transform or len(self._condition_errors) != 0 or\
                                 len(self._match_errors) != 0:
            return

        for handler, argument in self.transform.items():
            self.value = handler(
                value=self.value, argument=argument, context=self.context
            ).handle()

__init__(value, type, setup=None, conditions=None, matches=None, transform=None, optional=None, context=None, metadata=None)

Initializes the Pipe with a value, type, and processing configurations.

Parameters:

Name Type Description Default
value V

The value to process.

 required
type T

The expected type of the value (e.g., str, int).

 required
setup Optional[PipeTransform]

A dictionary of transform handlers and their arguments. Used for data setup (e.g. Strip). Use with caution. Setup runs after type validation, but before conditions, matches, and transform handlers. Only the value type is validated at the time of setup execution.

 None
conditions Optional[PipeConditions]

A dictionary of condition handlers and their arguments. Used for logical validation (e.g., MinLength, Equal).

 None
matches Optional[PipeMatches]

A dictionary of match handlers and their arguments. Used for pattern matching (e.g., Email, Regex).

 None
transform Optional[PipeTransform]

A dictionary of transform handlers and their arguments. Used for data modification (e.g., Strip, Capitalize).

 None
optional Optional[bool]

If True, the pipe is skipped if the value is falsy.

 None
context Optional[PipeContext]

Additional context for the handlers, typically the entire data dictionary being processed.

 None
metadata Optional[PipeMetadata]

Metadata about the pipe execution.

 None
Source code in pipeline/core/pipe/pipe.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(
    self,
    value: V,
    type: T,
    setup: Optional[PipeTransform] = None,
    conditions: Optional[PipeConditions] = None,
    matches: Optional[PipeMatches] = None,
    transform: Optional[PipeTransform] = None,
    optional: Optional[bool] = None,
    context: Optional[PipeContext] = None,
    metadata: Optional[PipeMetadata] = None
) -> None:
    """
    Initializes the Pipe with a value, type, and processing configurations.

    Args:
        value (V): The value to process.
        type (T): The expected type of the value (e.g., `str`, `int`).
        setup (Optional[PipeTransform]): A dictionary of transform handlers and their arguments.
            Used for data setup (e.g. Strip). Use with caution. Setup runs after type validation, but
            before conditions, matches, and transform handlers. Only the value type is
            validated at the time of setup execution.
        conditions (Optional[PipeConditions]): A dictionary of condition handlers and their arguments.
            Used for logical validation (e.g., `MinLength`, `Equal`).
        matches (Optional[PipeMatches]): A dictionary of match handlers and their arguments.
            Used for pattern matching (e.g., `Email`, `Regex`).
        transform (Optional[PipeTransform]): A dictionary of transform handlers and their arguments.
            Used for data modification (e.g., `Strip`, `Capitalize`).
        optional (Optional[bool]): If True, the pipe is skipped if the value is falsy.
        context (Optional[PipeContext]): Additional context for the handlers, typically the
            entire data dictionary being processed.
        metadata (Optional[PipeMetadata]): Metadata about the pipe execution.
    """
    self.value: V = value

    self.type: T = type

    self.setup: Optional[PipeTransform] = setup

    self.conditions: Optional[PipeConditions] = conditions
    self.matches: Optional[PipeMatches] = matches
    self.transform: Optional[PipeTransform] = transform

    self.optional: Optional[bool] = optional

    self.context: Optional[PipeContext] = context
    self.metadata: Optional[PipeMetadata] = metadata

    self._condition_errors: ConditionErrors = []
    self._match_errors: ConditionErrors = []

run()

Executes the pipe processing logic.

The method orchestrates the execution flow by delegating to specialized helper methods: 1. Checks if validation should be skipped (optional pipe with falsy value) 2. Validates the value type 3. Processes setup handlers (if any) 4. Processes condition handlers 5. Processes match handlers (only if no condition errors) 6. Processes transform handlers (only if no condition or match errors)

Note that transformations are ONLY applied if all validations (Conditions and Matches) pass. This provides a safe way to transform data, ensuring it is valid first.

Returns:

Type Description
PipeResult

PipeResult[V]: The result containing the processed value (or original value if errors occurred)
PipeResult

and lists of any condition or match errors.
Source code in pipeline/core/pipe/pipe.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def run(self) -> PipeResult:
    """
    Executes the pipe processing logic.

    The method orchestrates the execution flow by delegating to specialized helper methods:
    1. Checks if validation should be skipped (optional pipe with falsy value)
    2. Validates the value type
    3. Processes setup handlers (if any)
    4. Processes condition handlers
    5. Processes match handlers (only if no condition errors)
    6. Processes transform handlers (only if no condition or match errors)

    Note that transformations are ONLY applied if all validations (Conditions and Matches) pass.
    This provides a safe way to transform data, ensuring it is valid first.

    Returns:
        PipeResult[V]: The result containing the processed value (or original value if errors occurred)
        and lists of any condition or match errors.
    """
    if self._should_skip_validation():
        return self._construct_result()

    if not self._is_value_type_correct():
        return self._construct_result()

    self._process_setup()

    self._process_conditions()
    self._process_matches()
    self._process_transform()

    return self._construct_result()

pipeline.core.pipeline.pipeline

Pipeline

The Pipeline class orchestrates the execution of multiple pipes on a data dictionary.

It allows defining a sequence of processing steps (pipes) for specific fields in the input data. The pipeline iterates over the configuration, applies the corresponding pipe to each field, and aggregates any errors encountered. Hooks can be registered to execute before and after each pipe, allowing for side effects or custom logic. A custom error handler can also be provided to process aggregated errors.

Attributes:

Name Type Description
global_pre_hook ClassVar[PipelineHookFunc | None]

A function to be called before each pipe execution.
global_post_hook ClassVar[PipelineHookFunc | None]

A function to be called after each pipe execution.
global_teardown ClassVar[PipelineTeardownFunc | None]

A function that will run after the pipeline finishes execution, even if it failed. This runs before handle_errors.
global_handle_errors ClassVar[PipelineHandleErrorsFunc | None]

A function to handle errors collected during pipeline execution. This could be used to raise exceptions, log errors, or format them for a response.
Source code in pipeline/core/pipeline/pipeline.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
class Pipeline:
    """
    The Pipeline class orchestrates the execution of multiple pipes on a data dictionary.

    It allows defining a sequence of processing steps (pipes) for specific fields in the input data.
    The pipeline iterates over the configuration, applies the corresponding pipe to each field,
    and aggregates any errors encountered.
    Hooks can be registered to execute before and after each pipe, allowing for side effects or
    custom logic. A custom error handler can also be provided to process aggregated errors.

    Attributes:
        global_pre_hook (ClassVar[PipelineHookFunc | None]): A function to be called before each pipe execution.
        global_post_hook (ClassVar[PipelineHookFunc | None]): A function to be called after each pipe execution.
        global_teardown (ClassVar[PipelineTeardownFunc | None]): A function that will run after the 
            pipeline finishes execution, even if it failed. This runs before handle_errors.
        global_handle_errors (ClassVar[PipelineHandleErrorsFunc | None]): A function to handle
            errors collected during pipeline execution. This could be used to raise exceptions,
            log errors, or format them for a response.
    """
    global_pre_hook: ClassVar[PipelineHookFunc | None] = None
    """A function to be called before each pipe execution."""

    global_post_hook: ClassVar[PipelineHookFunc | None] = None
    """A function to be called after each pipe execution."""

    global_teardown: ClassVar[PipelineTeardownFunc | None] = None
    """A function that will run after the pipeline finishes execution, even if it failed."""

    global_handle_errors: ClassVar[PipelineHandleErrorsFunc | None] = None
    """A function to handle errors collected during pipeline execution."""
    def __init__(
        self,
        pre_hook: PipelineHookFunc | None = None,
        post_hook: PipelineHookFunc | None = None,
        teardown: PipelineTeardownFunc | None = None,
        handle_errors: PipelineHandleErrorsFunc | None = None,
        **pipes_config: PipelinePipeConfig
    ) -> None:
        """
        Initializes the Pipeline with a configuration of pipes.

        The `pipes_config` defines the schema and validation rules for the data. Each key in
        `pipes_config` corresponds to a key in the input data dictionary. The value is a
        dictionary of arguments required to initialize a `Pipe` instance (e.g., `type`,
        `conditions`, `matches`).

        Args:
            pre_hook (PipelineHookFunc | None): A function to be called before each pipe execution.
                The global_pre_hook will not run if a local pre_hook is defined.
            post_hook (PipelineHookFunc | None): A function to be called after each pipe execution.
                The global_post_hook will not run if a local pre_hook is defined.
            teardown (PipelineTeardownFunc | None): A function that will run after the entire 
                pipeline finishes execution. This runs before handle_errors and will execute 
                even if the pipeline failed. The global_teardown will not run if a local 
                teardown is defined.
            handle_errors (PipelineHandleErrorsFunc | None): A function to handle
                errors collected during pipeline execution. This could be used to raise exceptions,
                log errors, or format them for a response. The global_handle_errors will not run
                if a local handle_errors is defined.
            **pipes_config (PipelinePipeConfig): Configuration for the pipes.
                Keys represent the fields in the data dictionary to be processed,
                and values are the configuration for the corresponding pipe.
        """
        self.pre_hook: PipelineHookFunc | None = pre_hook
        self.post_hook: PipelineHookFunc | None = post_hook

        self.handle_errors: PipelineHandleErrorsFunc | None = handle_errors

        self.teardown: PipelineTeardownFunc | None = teardown

        self.pipes_config: dict[str, PipelinePipeConfig] = pipes_config

        self._errors: dict[str, ConditionErrors] = {}
        self._processed_data: dict = {}

        self._ran_before: bool = False

    def run(self, data: dict) -> PipelineResult:
        """
        Runs the pipeline on the provided data.

        This method iterates through the `pipes_config`. For each field, it executes
        the internal `self._process_field_pipe` method.

        Once the field processing is finished, if `teardown` is defined, it will run 
        before `handle_errors` and will execute even if the pipeline failed.

        After teardown, if `handle_errors` is defined, it is called with the
        collected errors.

        Args:
            data (dict): The input data dictionary to process. The dictionary may be modified in-place
                with transformed values.

        Returns:
            PipelineResult: A namedtuple containing the fields errors and processed_data. 
                The processed_data field contains the final, trustworthy data and will be `None` 
                if there are errors.
        """
        if self._ran_before:
            self._reset_state()

        self._ran_before = True

        context: PipeContext = data

        for field, pipe_config in self.pipes_config.items():
            self._process_field_pipe(
                data=data,
                context=context,
                field=field,
                pipe_config=pipe_config
            )

        if self.teardown:
            self.teardown(self)
        elif self.__class__.global_teardown:
            self.__class__.global_teardown(self)

        if self._errors:
            error_handler = self.handle_errors or self.__class__.global_handle_errors

            if error_handler:
                error_handler(self._errors)

        return PipelineResult(
            errors=self._errors or None,
            processed_data=None if self._errors else self._processed_data
        )

    def _process_field_pipe(
        self, data: dict, context: PipeContext, field: str,
        pipe_config: PipelinePipeConfig
    ) -> None:
        """
        Internal function that runs the pipe on a specific field and manages hooks.

        The execution flow is as follows:
        1. Value extraction: Retrieves the initial value from the input data.
        2. Hook preparation: Defines a closure-based Value class using nonlocal 
           to allow hooks to get or set the variable directly in the local scope.
        3. Pre-hook execution: Runs the local pre_hook if defined. Otherwise, 
           falls back to the Pipeline.global_pre_hook.
        4. Context Management: If the current value is a dictionary, it overrides 
           the context for the downstream pipe execution.
        5. Pipe execution: Initializes and runs a Pipe instance to handle 
           validation, matching, and transformation.
        6. Error handling: Checks for condition_errors or match_errors. If found, 
           the field is marked invalid and errors are stored in self._errors.
        7. Post-Hook Execution: Updates the hook's is_valid state and executes 
           the local or global post-hook.
        8. Data Persistence: Saves the final value to self._processed_data.

        Args:
            data: The source dictionary containing the field value.
            context: The shared context for the pipeline execution.
            field: The name of the field being processed.
            pipe_config: Configuration parameters for the specific pipe.
        """
        from pipeline.core.pipe.pipe import Pipe

        value: Any = data.get(field, None)

        class Value:
            @property
            def get(self) -> Any:
                nonlocal value

                return value

            def set(self, new_value: Any) -> Any:
                nonlocal value

                value = new_value

                return self.get

        hook = PipelineHook(
            field=field, value=Value(), is_valid=None, pipe_config=pipe_config
        )

        if self.pre_hook:
            self.pre_hook(hook)
        elif self.__class__.global_pre_hook:
            self.__class__.global_pre_hook(hook)

        if isinstance(value, dict):
            context = value

        pipe: Pipe = Pipe(value=value, **pipe_config, context=context)

        value, condition_errors, match_errors = pipe.run()

        is_valid: bool = not bool(condition_errors or match_errors)

        if not is_valid:
            self._errors[field] = [*condition_errors, *match_errors]

        hook.is_valid = is_valid

        if self.post_hook:
            self.post_hook(hook)
        elif self.__class__.global_post_hook:
            self.__class__.global_post_hook(hook)

        self._processed_data[field] = value

    def _reset_state(self):
        """
        Resets internal state variables and execution flags.

        Clears the error and processed data and resets the execution tracker to 
        its initial state.
        """
        self._errors = {}
        self._processed_data = {}

        self._ran_before = False

    def __call__(self, func: Callable[P, F]) -> Callable[P, F]:
        """
        Decorator to run the pipeline before a function execution.

        When the decorated function is called, the pipeline is run using the function's
        keyword arguments (`kwargs`) as the input data. If validation fails, and no
        `handle_errors` is provided, a `PipelineException` is raised.

        Args:
            func (Any): The function to be decorated. The pipeline will run on the
                keyword arguments passed to this function.

        Returns:
            Any: The wrapped function.

        Raises:
            PipelineException: If the pipeline encounters errors and no error handler is defined.
        """
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> F:
            result: PipelineResult = self.run(data=kwargs)

            if result.errors and not self.handle_errors or result.processed_data is None:
                raise PipelineException(result.errors)

            updated_kwargs: dict = {**kwargs, **result.processed_data}

            return func(*args, **updated_kwargs)

        return wrapper

global_handle_errors = None class-attribute

A function to handle errors collected during pipeline execution.

global_post_hook = None class-attribute

A function to be called after each pipe execution.

global_pre_hook = None class-attribute

A function to be called before each pipe execution.

global_teardown = None class-attribute

A function that will run after the pipeline finishes execution, even if it failed.

__call__(func)

Decorator to run the pipeline before a function execution.

When the decorated function is called, the pipeline is run using the function's keyword arguments (kwargs) as the input data. If validation fails, and no handle_errors is provided, a PipelineException is raised.

Parameters:

Name Type Description Default
func Any

The function to be decorated. The pipeline will run on the keyword arguments passed to this function.

 required

Returns:

Name Type Description
Any Callable[P, F]

The wrapped function.

Raises:

Type Description
PipelineException

If the pipeline encounters errors and no error handler is defined.
Source code in pipeline/core/pipeline/pipeline.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def __call__(self, func: Callable[P, F]) -> Callable[P, F]:
    """
    Decorator to run the pipeline before a function execution.

    When the decorated function is called, the pipeline is run using the function's
    keyword arguments (`kwargs`) as the input data. If validation fails, and no
    `handle_errors` is provided, a `PipelineException` is raised.

    Args:
        func (Any): The function to be decorated. The pipeline will run on the
            keyword arguments passed to this function.

    Returns:
        Any: The wrapped function.

    Raises:
        PipelineException: If the pipeline encounters errors and no error handler is defined.
    """
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> F:
        result: PipelineResult = self.run(data=kwargs)

        if result.errors and not self.handle_errors or result.processed_data is None:
            raise PipelineException(result.errors)

        updated_kwargs: dict = {**kwargs, **result.processed_data}

        return func(*args, **updated_kwargs)

    return wrapper

__init__(pre_hook=None, post_hook=None, teardown=None, handle_errors=None, **pipes_config)

Initializes the Pipeline with a configuration of pipes.

The pipes_config defines the schema and validation rules for the data. Each key in pipes_config corresponds to a key in the input data dictionary. The value is a dictionary of arguments required to initialize a Pipe instance (e.g., type, conditions, matches).

Parameters:

Name Type Description Default
pre_hook PipelineHookFunc | None

A function to be called before each pipe execution. The global_pre_hook will not run if a local pre_hook is defined.

 None
post_hook PipelineHookFunc | None

A function to be called after each pipe execution. The global_post_hook will not run if a local pre_hook is defined.

 None
teardown PipelineTeardownFunc | None

A function that will run after the entire pipeline finishes execution. This runs before handle_errors and will execute even if the pipeline failed. The global_teardown will not run if a local teardown is defined.

 None
handle_errors PipelineHandleErrorsFunc | None

A function to handle errors collected during pipeline execution. This could be used to raise exceptions, log errors, or format them for a response. The global_handle_errors will not run if a local handle_errors is defined.

 None
**pipes_config PipelinePipeConfig

Configuration for the pipes. Keys represent the fields in the data dictionary to be processed, and values are the configuration for the corresponding pipe.

 {}
Source code in pipeline/core/pipeline/pipeline.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(
    self,
    pre_hook: PipelineHookFunc | None = None,
    post_hook: PipelineHookFunc | None = None,
    teardown: PipelineTeardownFunc | None = None,
    handle_errors: PipelineHandleErrorsFunc | None = None,
    **pipes_config: PipelinePipeConfig
) -> None:
    """
    Initializes the Pipeline with a configuration of pipes.

    The `pipes_config` defines the schema and validation rules for the data. Each key in
    `pipes_config` corresponds to a key in the input data dictionary. The value is a
    dictionary of arguments required to initialize a `Pipe` instance (e.g., `type`,
    `conditions`, `matches`).

    Args:
        pre_hook (PipelineHookFunc | None): A function to be called before each pipe execution.
            The global_pre_hook will not run if a local pre_hook is defined.
        post_hook (PipelineHookFunc | None): A function to be called after each pipe execution.
            The global_post_hook will not run if a local pre_hook is defined.
        teardown (PipelineTeardownFunc | None): A function that will run after the entire 
            pipeline finishes execution. This runs before handle_errors and will execute 
            even if the pipeline failed. The global_teardown will not run if a local 
            teardown is defined.
        handle_errors (PipelineHandleErrorsFunc | None): A function to handle
            errors collected during pipeline execution. This could be used to raise exceptions,
            log errors, or format them for a response. The global_handle_errors will not run
            if a local handle_errors is defined.
        **pipes_config (PipelinePipeConfig): Configuration for the pipes.
            Keys represent the fields in the data dictionary to be processed,
            and values are the configuration for the corresponding pipe.
    """
    self.pre_hook: PipelineHookFunc | None = pre_hook
    self.post_hook: PipelineHookFunc | None = post_hook

    self.handle_errors: PipelineHandleErrorsFunc | None = handle_errors

    self.teardown: PipelineTeardownFunc | None = teardown

    self.pipes_config: dict[str, PipelinePipeConfig] = pipes_config

    self._errors: dict[str, ConditionErrors] = {}
    self._processed_data: dict = {}

    self._ran_before: bool = False

run(data)

Runs the pipeline on the provided data.

This method iterates through the pipes_config. For each field, it executes the internal self._process_field_pipe method.

Once the field processing is finished, if teardown is defined, it will run before handle_errors and will execute even if the pipeline failed.

After teardown, if handle_errors is defined, it is called with the collected errors.

Parameters:

Name Type Description Default
data dict

The input data dictionary to process. The dictionary may be modified in-place with transformed values.

 required

Returns:

Name Type Description
PipelineResult PipelineResult

A namedtuple containing the fields errors and processed_data. The processed_data field contains the final, trustworthy data and will be None if there are errors.
Source code in pipeline/core/pipeline/pipeline.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def run(self, data: dict) -> PipelineResult:
    """
    Runs the pipeline on the provided data.

    This method iterates through the `pipes_config`. For each field, it executes
    the internal `self._process_field_pipe` method.

    Once the field processing is finished, if `teardown` is defined, it will run 
    before `handle_errors` and will execute even if the pipeline failed.

    After teardown, if `handle_errors` is defined, it is called with the
    collected errors.

    Args:
        data (dict): The input data dictionary to process. The dictionary may be modified in-place
            with transformed values.

    Returns:
        PipelineResult: A namedtuple containing the fields errors and processed_data. 
            The processed_data field contains the final, trustworthy data and will be `None` 
            if there are errors.
    """
    if self._ran_before:
        self._reset_state()

    self._ran_before = True

    context: PipeContext = data

    for field, pipe_config in self.pipes_config.items():
        self._process_field_pipe(
            data=data,
            context=context,
            field=field,
            pipe_config=pipe_config
        )

    if self.teardown:
        self.teardown(self)
    elif self.__class__.global_teardown:
        self.__class__.global_teardown(self)

    if self._errors:
        error_handler = self.handle_errors or self.__class__.global_handle_errors

        if error_handler:
            error_handler(self._errors)

    return PipelineResult(
        errors=self._errors or None,
        processed_data=None if self._errors else self._processed_data
    )

pipeline.handlers.base_handler.base_handler

BaseHandler

Bases: ABC, Generic[V, A]

Abstract base class for all handlers in the pipeline.

Handlers are the building blocks of the pipeline, responsible for processing values (validation, matching, transformation) based on provided arguments and context. They support different modes of operation to handle single values, items in a collection, or values dependent on other context fields.

Attributes:

Name Type Description
FLAGS ClassVar[tuple[Flag, ...]]

Flags acting as settings for the handler. Example: ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR stops processing if the handler fails.
SUPPORT ClassVar[tuple[HandlerMode, ...]]

Supported handler modes. - HandlerMode.ROOT: The handler processes the value directly. - HandlerMode.ITEM: The handler processes each item in a list/dict. - HandlerMode.CONTEXT: The handler uses another field from the context as an argument.
CONTEXT_ARGUMENT_BUILDER ClassVar[Optional[Callable]]

Helper to build arguments from context. Used in CONTEXT mode to transform the context value before using it as an argument.
Source code in pipeline/handlers/base_handler/base_handler.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
class BaseHandler(ABC, Generic[V, A]):
    """
    Abstract base class for all handlers in the pipeline.

    Handlers are the building blocks of the pipeline, responsible for processing values
    (validation, matching, transformation) based on provided arguments and context.
    They support different modes of operation to handle single values, items in a collection,
    or values dependent on other context fields.

    Attributes:
        FLAGS (ClassVar[tuple[Flag, ...]]): Flags acting as settings for the handler.
            Example: `ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR` stops processing if the handler fails.
        SUPPORT (ClassVar[tuple[HandlerMode, ...]]): Supported handler modes.
            - `HandlerMode.ROOT`: The handler processes the value directly.
            - `HandlerMode.ITEM`: The handler processes each item in a list/dict.
            - `HandlerMode.CONTEXT`: The handler uses another field from the context as an argument.
        CONTEXT_ARGUMENT_BUILDER (ClassVar[Optional[Callable]]): Helper to build arguments from context.
            Used in CONTEXT mode to transform the context value before using it as an argument.
    """
    FLAGS: ClassVar[tuple[Flag, ...]] = tuple()

    SUPPORT: ClassVar[tuple[HandlerMode, ...]] = tuple()

    CONTEXT_ARGUMENT_BUILDER: ClassVar[Optional[Callable[['BaseHandler', Any],
                                                         Any]]] = None

    def __init__(
        self,
        value: V,
        argument: A,
        context: Optional[PipeContext] = None,
        metadata: Optional[PipeMetadata] = None,
        _mode: HandlerMode = HandlerMode.ROOT,
        _item_use_key: Optional[bool] = False,
        _preferred_value_type: Optional[type] = None
    ) -> None:
        """
        Initializes the BaseHandler.

        Args:
            value (V): The value to process.
            argument (A): The argument for the handler.
            context (Optional[PipeContext]): Additional context for the handler.
            metadata (Optional[PipeMetadata]): Metadata about the pipe execution.
            _mode (HandlerMode): The mode in which the handler is operating.
            _item_use_key (Optional[bool]): If True and in ITEM mode, the handler operates on the
                key of a dictionary item instead of the value.
            _preferred_value_type (Optional[type]): Specific type to prefer/enforce during type validation.
        """
        self.value: V = value
        self.argument: A = argument

        self.input_value: V = value
        self.input_argument: A = argument

        self.context: PipeContext = context or {}
        self.metadata: PipeMetadata = metadata or {}

        self._mode: HandlerMode = _mode

        self._item_index: Optional[int | str] = None
        self._item_use_key: Optional[bool] = _item_use_key

        self._preferred_value_type: Optional[type] = _preferred_value_type

        self._prepare_and_validate_handler()

    def handle(self) -> Any:
        """
        Executes the handler logic based on the current mode.

        It delegates to `_handle()` for ROOT and CONTEXT modes, and `_handle_item_mode()`
        for ITEM mode.

        Returns:
            Any: The result of the handling operation. The return type depends on the specific
            handler implementation (e.g., specific error type, boolean, or transformed value).

        Raises:
            HandlerException: If the handler mode is invalid.
        """
        if self._mode in (HandlerMode.ROOT, HandlerMode.CONTEXT):
            return self._handle()
        elif self._mode == HandlerMode.ITEM:
            return self._handle_item_mode()
        else:
            raise HandlerException("Invalid handler mode.")

    def _prepare_and_validate_handler(self) -> None:
        """
        Prepares the handler for the current mode and validates input types.

        This methods checks if the requested mode is supported, perpares the handler argument
        (especially for CONTEXT mode), and validates that value and argument types match
        expectations (generics).

        Raises:
            HandlerModeUnsupported: If the current mode is not supported by the handler.
        """
        if self._mode not in self.SUPPORT:
            raise HandlerModeUnsupported(handler_mode=self._mode)

        self._prepare_handler_for_mode()
        self._validate_type_if_possible()

    def _prepare_handler_for_mode(self) -> None:
        """
        Performs specific preparation steps based on the handler mode.

        For CONTEXT mode, it retrieves the argument from the context using the provided
        argument name (stored in `self.argument`). It also handles optional argument transformation
        via `CONTEXT_ARGUMENT_BUILDER`.
        """
        match self._mode:
            case HandlerMode.CONTEXT:
                context_value: Any = self.context.get(str(self.argument), None)

                if context_value is None:
                    raise HandlerModeMissingContextValue(
                        argument=str(self.argument)
                    )

                self.argument = self.CONTEXT_ARGUMENT_BUILDER(
                    context_value
                ) if self.CONTEXT_ARGUMENT_BUILDER else context_value

    def _is_valid_type(
        self, value: Any, expected_type: type | tuple[type, ...]
    ) -> bool:
        """
        Checks if a value matches the expected type(s).

        Args:
            value (Any): The value to check.
            expected_type (type | tuple[type, ...]): The expected type or tuple of types.

        Returns:
            bool: True if the value matches the expected type, False otherwise.
        """
        if isinstance(expected_type, Iterable):
            if Any in expected_type:
                return True

            return isinstance(value, expected_type)

        return expected_type == Any or isinstance(value, expected_type)

    def _validate_type_if_possible(self) -> None:
        """
        Validates the types of the input value and argument against the class generics.

        This ensures type safety at runtime, verifying that the handler is being applied
        to compatible data.

        Value type is only verified for ROOT and CONTEXT modes.For ITEM mode, the handler
        must implement its own logic.

        Raises:
            HandlerInvalidValueType: If the value type is invalid.
            HandlerInvalidArgumentType: If the argument type is invalid.
        """
        if self._mode in (HandlerMode.ROOT, HandlerMode.CONTEXT):
            if not self._is_valid_type(self.value, self._expected_value_type):
                raise HandlerInvalidValueType(handler=self)

        if not self._is_valid_type(self.argument, self._expected_argument_type):
            raise HandlerInvalidArgumentType(handler=self)

    @abstractmethod
    def _handle(self) -> Any:
        """
        Abstract method to implement the main handling logic.
        """
        ...

    @abstractmethod
    def _handle_item_mode(self) -> Any:
        """
        Abstract method to implement the handling logic for ITEM mode.
        """
        ...

    @property
    def id(self) -> str:
        """
        Returns a unique identifier for the handler based on its class name.

        Returns:
            str: The snake_case identifier of the handler (e.g., 'MaxLength' -> 'max_length').
        """
        s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', self.__class__.__name__)

        return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()

    @cached_property
    def _expected_types(self) -> HandlerExpectedTypes:
        """
        Determines the expected value and argument types from the generic type arguments.

        This introspects the class definition to find the concrete types provided for
        `BaseHandler[V, A]`.

        Returns:
            HandlerExpectedTypes: Named tuple containing expected value and argument types.

        Raises:
            TypeError: If the class does not inherit from BaseHandler correctly or has incorrect generic arguments.
            HandlerInvalidPreferredValueType: If the preferred value type is invalid.
        """
        orig_bases: list[Any] = getattr(self, "__orig_bases__", [])

        if len(orig_bases) != 1:
            raise TypeError(
                "Handler subclass must inherit from a generic base class (e.g. BaseHandler[V, A])."
            )

        generic_args: tuple[Any, ...] = get_args(orig_bases[0])

        if len(generic_args) != 2:
            raise TypeError(
                f"Expected 2 generic arguments on base class, found {len(generic_args)}."
            )

        def _unpack_generic_args(arg: Any) -> Any:
            generic_args = get_args(arg)

            if generic_args:
                return tuple(
                    _unpack_generic_args(sub_arg) for sub_arg in generic_args
                )

            return arg

        expected_value_type: tuple[type, ...] | type = _unpack_generic_args(
            generic_args[0]
        )

        expected_argument_type: tuple[type, ...] | type = _unpack_generic_args(
            generic_args[1]
        )

        if not isinstance(expected_value_type, tuple):
            expected_value_types = (expected_value_type, )
        else:
            expected_value_types = expected_value_type

        if not isinstance(expected_argument_type, tuple):
            expected_argument_type = (expected_argument_type, )
        else:
            expected_argument_type = expected_argument_type

        if self._preferred_value_type:
            if self._preferred_value_type not in expected_value_types and Any not in expected_value_types:
                raise HandlerInvalidPreferredValueType(
                    self, expected_value_type=expected_value_types
                )

            return HandlerExpectedTypes(
                value=(self._preferred_value_type, ),
                argument=expected_argument_type
            )

        return HandlerExpectedTypes(
            value=expected_value_types, argument=expected_argument_type
        )

    @property
    def _expected_value_type(self) -> tuple[type, ...]:
        return self._expected_types.value

    @property
    def _expected_argument_type(self) -> tuple[type, ...]:
        return self._expected_types.argument

id property

Returns a unique identifier for the handler based on its class name.

Returns:

Name Type Description
str str

The snake_case identifier of the handler (e.g., 'MaxLength' -> 'max_length').

__init__(value, argument, context=None, metadata=None, _mode=HandlerMode.ROOT, _item_use_key=False, _preferred_value_type=None)

Initializes the BaseHandler.

Parameters:

Name Type Description Default
value V

The value to process.

 required
argument A

The argument for the handler.

 required
context Optional[PipeContext]

Additional context for the handler.

 None
metadata Optional[PipeMetadata]

Metadata about the pipe execution.

 None
_mode HandlerMode

The mode in which the handler is operating.

 ROOT
_item_use_key Optional[bool]

If True and in ITEM mode, the handler operates on the key of a dictionary item instead of the value.

 False
_preferred_value_type Optional[type]

Specific type to prefer/enforce during type validation.

 None
Source code in pipeline/handlers/base_handler/base_handler.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(
    self,
    value: V,
    argument: A,
    context: Optional[PipeContext] = None,
    metadata: Optional[PipeMetadata] = None,
    _mode: HandlerMode = HandlerMode.ROOT,
    _item_use_key: Optional[bool] = False,
    _preferred_value_type: Optional[type] = None
) -> None:
    """
    Initializes the BaseHandler.

    Args:
        value (V): The value to process.
        argument (A): The argument for the handler.
        context (Optional[PipeContext]): Additional context for the handler.
        metadata (Optional[PipeMetadata]): Metadata about the pipe execution.
        _mode (HandlerMode): The mode in which the handler is operating.
        _item_use_key (Optional[bool]): If True and in ITEM mode, the handler operates on the
            key of a dictionary item instead of the value.
        _preferred_value_type (Optional[type]): Specific type to prefer/enforce during type validation.
    """
    self.value: V = value
    self.argument: A = argument

    self.input_value: V = value
    self.input_argument: A = argument

    self.context: PipeContext = context or {}
    self.metadata: PipeMetadata = metadata or {}

    self._mode: HandlerMode = _mode

    self._item_index: Optional[int | str] = None
    self._item_use_key: Optional[bool] = _item_use_key

    self._preferred_value_type: Optional[type] = _preferred_value_type

    self._prepare_and_validate_handler()

handle()

Executes the handler logic based on the current mode.

It delegates to _handle() for ROOT and CONTEXT modes, and _handle_item_mode() for ITEM mode.

Returns:

Name Type Description
Any Any

The result of the handling operation. The return type depends on the specific
Any

handler implementation (e.g., specific error type, boolean, or transformed value).

Raises:

Type Description
HandlerException

If the handler mode is invalid.
Source code in pipeline/handlers/base_handler/base_handler.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def handle(self) -> Any:
    """
    Executes the handler logic based on the current mode.

    It delegates to `_handle()` for ROOT and CONTEXT modes, and `_handle_item_mode()`
    for ITEM mode.

    Returns:
        Any: The result of the handling operation. The return type depends on the specific
        handler implementation (e.g., specific error type, boolean, or transformed value).

    Raises:
        HandlerException: If the handler mode is invalid.
    """
    if self._mode in (HandlerMode.ROOT, HandlerMode.CONTEXT):
        return self._handle()
    elif self._mode == HandlerMode.ITEM:
        return self._handle_item_mode()
    else:
        raise HandlerException("Invalid handler mode.")

pipeline.handlers.base_handler.handler_modifiers

Context(handler)

Modifier ensure the handler is run in CONTEXT mode.

In CONTEXT mode, the handler's argument is retrieved from the pipeline context using the provided argument as a key.

Parameters:

Name Type Description Default
handler Type[T]

The handler class to modify.

 required

Returns:

Name Type Description
partial

A partial application of the handler with _mode=HandlerMode.CONTEXT.
Source code in pipeline/handlers/base_handler/handler_modifiers.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def Context(handler: Type[T]):
    """
    Modifier ensure the handler is run in CONTEXT mode.

    In CONTEXT mode, the handler's argument is retrieved from the pipeline context
    using the provided argument as a key.

    Args:
        handler (Type[T]): The handler class to modify.

    Returns:
        partial: A partial application of the handler with _mode=HandlerMode.CONTEXT.
    """
    return partial(handler, _mode=HandlerMode.CONTEXT)

Item(handler, use_key=False, only_consider=None)

Modifier to ensure the handler is run in ITEM mode.

In ITEM mode, the handler is applied to each item in an iterable.

Parameters:

Name Type Description Default
handler Type[T] | partial[T]

The handler class or partial to modify.

 required
use_key Optional[bool]

If True, the handler uses the item's key (e.g., in a dictionary) instead of value.

 False
only_consider Optional[type]

Specific type to filter items for processing.

 None

Returns:

Name Type Description
partial

A partial application of the handler with ITEM mode settings.
Source code in pipeline/handlers/base_handler/handler_modifiers.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def Item(
    handler: Type[T] | partial[T],
    use_key: Optional[bool] = False,
    only_consider: Optional[type] = None
):
    """
    Modifier to ensure the handler is run in ITEM mode.

    In ITEM mode, the handler is applied to each item in an iterable.

    Args:
        handler (Type[T] | partial[T]): The handler class or partial to modify.
        use_key (Optional[bool]): If True, the handler uses the item's key (e.g., in a dictionary) instead of value.
        only_consider (Optional[type]): Specific type to filter items for processing.

    Returns:
        partial: A partial application of the handler with ITEM mode settings.
    """
    return partial(
        handler,
        _mode=HandlerMode.ITEM,
        _item_use_key=use_key,
        _preferred_value_type=only_consider
    )

pipeline.handlers.condition_handler.condition_handler

ConditionHandler

Bases: BaseHandler[V, A]

Abstract base class for specific condition implementations.

This class provides the infrastructure for condition checking, including error message generation and support for different handling modes (ROOT, ITEM). It expects subclasses to implement the query method.

Source code in pipeline/handlers/condition_handler/condition_handler.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class ConditionHandler(BaseHandler[V, A]):
    """
    Abstract base class for specific condition implementations.

    This class provides the infrastructure for condition checking, including error message generation
    and support for different handling modes (ROOT, ITEM).
    It expects subclasses to implement the `query` method.
    """
    ERROR_BUILDER: ClassVar[Callable[['ConditionHandler'],
                                     ConditionError]] = default_error_builder

    ERROR_TEMPLATES: ClassVar[ConditionErrorTemplates]

    def __init__(
        self,
        value: V,
        argument: A,
        context: Optional[PipeContext] = None,
        metadata: Optional[PipeMetadata] = None,
        _mode: HandlerMode = HandlerMode.ROOT,
        _item_use_key: Optional[bool] = False,
        _preferred_value_type: Optional[type] = None
    ) -> None:
        """
        Initializes the ConditionHandler.

        It ensures that if ROOT mode is supported, a corresponding error template is present.
        """
        super().__init__(
            value, argument, context, metadata, _mode, _item_use_key,
            _preferred_value_type
        )

        if HandlerMode.ROOT in self.SUPPORT and HandlerMode.ROOT not in self.ERROR_TEMPLATES:
            raise ConditionMissingRootErrorMsg()

    @abstractmethod
    def query(self) -> bool:
        """
        Performs the condition check.

        Returns:
            bool: True if the condition is met, False otherwise.
        """
        ...

    def _handle(self) -> Optional[ConditionError]:
        """
        Handles the condition check in ROOT or CONTEXT mode.

        Returns:
            Optional[ConditionError]: An error object if the check fails, None otherwise.
        """
        if not self.query():
            return self.ERROR_BUILDER()

    def _handle_item_mode(self) -> Optional[dict[str | int, ConditionError]]:
        """
        Handles the condition check in ITEM mode (for iterables).

        Iterates over the input value and applies the check to each item.

        Returns:
            Optional[dict[str | int, ConditionError]]: A dictionary of errors keyed by item index/key,
            or None if no errors occurred.

        Raises:
            HandlerModeException: If the input value is not a supported iterable type.
        """
        errors = {}

        if isinstance(self.input_value, (list, tuple, set)):
            items = enumerate(self.input_value)
        elif isinstance(self.input_value, dict):
            items = self.input_value.items()
        else:
            raise HandlerModeException(
                "Cannot iterate over value. Expected a list, tuple, set, or dict."
            )

        for key, value in items:
            if self._item_use_key:
                value = key

            if not self._is_valid_type(value, self._expected_value_type):
                continue

            # NOTE: We use can cast() here because we checked if the value type is valid but linter does not know that.
            self.value = cast(V, value)

            self._item_index = key

            if not self.query():
                errors[key] = (self.ERROR_BUILDER())

        return errors if errors else None

    @property
    def error_msg(self) -> Any:
        """
        Generates the error message based on the current mode and error templates.

        Returns:
            Any: The generated error message.

        Raises:
            ConditionMissingRootErrorMsg: If the root error template is missing.
        """
        if self._mode in self.ERROR_TEMPLATES:
            return self.ERROR_TEMPLATES[self._mode](self)

        if HandlerMode.ROOT not in self.ERROR_TEMPLATES:
            raise ConditionMissingRootErrorMsg()

        return self.ERROR_TEMPLATES[HandlerMode.ROOT](self)

error_msg property

Generates the error message based on the current mode and error templates.

Returns:

Name Type Description
Any Any

The generated error message.

Raises:

Type Description
ConditionMissingRootErrorMsg

If the root error template is missing.

__init__(value, argument, context=None, metadata=None, _mode=HandlerMode.ROOT, _item_use_key=False, _preferred_value_type=None)

Initializes the ConditionHandler.

It ensures that if ROOT mode is supported, a corresponding error template is present.

Source code in pipeline/handlers/condition_handler/condition_handler.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def __init__(
    self,
    value: V,
    argument: A,
    context: Optional[PipeContext] = None,
    metadata: Optional[PipeMetadata] = None,
    _mode: HandlerMode = HandlerMode.ROOT,
    _item_use_key: Optional[bool] = False,
    _preferred_value_type: Optional[type] = None
) -> None:
    """
    Initializes the ConditionHandler.

    It ensures that if ROOT mode is supported, a corresponding error template is present.
    """
    super().__init__(
        value, argument, context, metadata, _mode, _item_use_key,
        _preferred_value_type
    )

    if HandlerMode.ROOT in self.SUPPORT and HandlerMode.ROOT not in self.ERROR_TEMPLATES:
        raise ConditionMissingRootErrorMsg()

query() abstractmethod

Performs the condition check.

Returns:

Name Type Description
bool bool

True if the condition is met, False otherwise.
Source code in pipeline/handlers/condition_handler/condition_handler.py 
60
61
62
63
64
65
66
67
68
@abstractmethod
def query(self) -> bool:
    """
    Performs the condition check.

    Returns:
        bool: True if the condition is met, False otherwise.
    """
    ...

pipeline.handlers.condition_handler.condition

Condition

Registry for all condition handlers.

This class groups all available condition handlers (e.g., ValueType, MinLength, Equal) for easy access.

Source code in pipeline/handlers/condition_handler/condition.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class Condition:
    """
    Registry for all condition handlers.

    This class groups all available condition handlers (e.g., ValueType, MinLength, Equal)
    for easy access.
    """
    class ValueType(ConditionHandler[Any, type]):
        """
        A built-in condition handler to validate the type of the value.

        This handler is automatically used by the Pipe to ensure the passed value matches
        the expected type defined in the Pipe.
        """
        FLAGS = (ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR, )

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self: f"Invalid type. Expected {self.argument.__name__}."
        }

        def query(self):
            return isinstance(self.value, self.argument)

    class MinLength(ConditionHandler[str | list | dict, int]):
        """Ensures the collection or string has at least N items/characters"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Too short. Minimum length is {self.argument} characters."
                if isinstance(self.value, str) else
                f"Too few items. Minimum count is {self.argument}."
        }

        def query(self):
            return len(self.value) >= self.argument

    class MaxLength(ConditionHandler[str | list | dict, int]):
        """Ensures the collection or string does not exceed N items/characters"""
        FLAGS = (ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR, )

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Too long. Maximum length is {self.argument} characters."
                if isinstance(self.value, str) else
                f"Too many items. Maximum count is {self.argument}."
        }

        def query(self):
            return len(self.value) <= self.argument

    class ExactLength(ConditionHandler[str | list | dict, int]):
        """Ensures the collection or string is exactly N items/characters long"""
        FLAGS = (ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR, )

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Invalid length. Must be exactly {self.argument} characters."
                if isinstance(self.value, str) else
                f"Invalid count. Must contain exactly {self.argument} items."
        }

        def query(self):
            return len(self.value) == self.argument

    class MinNumber(ConditionHandler[int | float, int | float]):
        """Ensures the numeric value is greater than or equal to N"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self: f"Value must be at least {self.argument}."
        }

        def query(self):
            return self.value >= self.argument

    class MaxNumber(ConditionHandler[int | float, int | float]):
        """Ensures the numeric value is less than or equal to N"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda self: f"Value must be at most {self.argument}."
        }

        def query(self):
            return self.value <= self.argument

    class IncludedIn(ConditionHandler[Any, Iterable]):
        """Ensures the value exists within the provided Iterable"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Selected option is invalid. Please choose from the {list(self.argument)}."
        }

        def query(self):
            return self.value in self.argument

    class NotIncludedIn(ConditionHandler[Any, Iterable]):
        """Ensures the value does not exist within the provided blacklist"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "This value is not allowed."
        }

        def query(self):
            return self.value not in self.argument

    class Equal(ConditionHandler[Any, Any]):
        """Ensures the value is strictly equal to the argument or a specific context field"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Value does not match the expected value."
        }

        def query(self):
            return self.value == self.argument

    class NotEqual(ConditionHandler[Any, Any]):
        """Ensures the value is strictly not equal to the argument or a specific context field"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Value must be different."
        }

        def query(self):
            return self.value != self.argument

    class MatchesField(ConditionHandler[Any, Any]):
        """Validates that the current value matches the value of another field in the context (e.g., password confirmation)"""
        SUPPORT = (HandlerMode.CONTEXT, )

        ERROR_TEMPLATES = {
            HandlerMode.CONTEXT:
                lambda self: f'Must match the "{self.input_argument}" field.'
        }

        def query(self):
            return self.value == self.argument

    class DoesNotMatchField(ConditionHandler[Any, Any]):
        """Validates that the current value does not match the value of another field in the context (e.g., new password != old password)"""
        SUPPORT = (HandlerMode.CONTEXT, )

        ERROR_TEMPLATES = {
            HandlerMode.CONTEXT:
                lambda self:
                f'Must be different from the "{self.input_argument}" field.'
        }

        def query(self):
            return self.value != self.argument

    class Pipeline(ConditionHandler[dict, Pipeline]):
        """Validates a dictionary using the same rules as the normal pipeline, but for nested data."""
        SUPPORT = (HandlerMode.ROOT, )

        ERROR_BUILDER = lambda self: self.metadata['errors']

        ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: None}

        def query(self):
            self.metadata['errors'] = self.argument.run(data=self.value).errors

            return self.metadata['errors'] is None

DoesNotMatchField

Bases: ConditionHandler[Any, Any]

Validates that the current value does not match the value of another field in the context (e.g., new password != old password)

Source code in pipeline/handlers/condition_handler/condition.py 
170
171
172
173
174
175
176
177
178
179
180
181
class DoesNotMatchField(ConditionHandler[Any, Any]):
    """Validates that the current value does not match the value of another field in the context (e.g., new password != old password)"""
    SUPPORT = (HandlerMode.CONTEXT, )

    ERROR_TEMPLATES = {
        HandlerMode.CONTEXT:
            lambda self:
            f'Must be different from the "{self.input_argument}" field.'
    }

    def query(self):
        return self.value != self.argument

Equal

Bases: ConditionHandler[Any, Any]

Ensures the value is strictly equal to the argument or a specific context field

Source code in pipeline/handlers/condition_handler/condition.py 
135
136
137
138
139
140
141
142
143
144
145
class Equal(ConditionHandler[Any, Any]):
    """Ensures the value is strictly equal to the argument or a specific context field"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Value does not match the expected value."
    }

    def query(self):
        return self.value == self.argument

ExactLength

Bases: ConditionHandler[str | list | dict, int]

Ensures the collection or string is exactly N items/characters long

Source code in pipeline/handlers/condition_handler/condition.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class ExactLength(ConditionHandler[str | list | dict, int]):
    """Ensures the collection or string is exactly N items/characters long"""
    FLAGS = (ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR, )

    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Invalid length. Must be exactly {self.argument} characters."
            if isinstance(self.value, str) else
            f"Invalid count. Must contain exactly {self.argument} items."
    }

    def query(self):
        return len(self.value) == self.argument

IncludedIn

Bases: ConditionHandler[Any, Iterable]

Ensures the value exists within the provided Iterable

Source code in pipeline/handlers/condition_handler/condition.py 
111
112
113
114
115
116
117
118
119
120
121
122
class IncludedIn(ConditionHandler[Any, Iterable]):
    """Ensures the value exists within the provided Iterable"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Selected option is invalid. Please choose from the {list(self.argument)}."
    }

    def query(self):
        return self.value in self.argument

MatchesField

Bases: ConditionHandler[Any, Any]

Validates that the current value matches the value of another field in the context (e.g., password confirmation)

Source code in pipeline/handlers/condition_handler/condition.py 
158
159
160
161
162
163
164
165
166
167
168
class MatchesField(ConditionHandler[Any, Any]):
    """Validates that the current value matches the value of another field in the context (e.g., password confirmation)"""
    SUPPORT = (HandlerMode.CONTEXT, )

    ERROR_TEMPLATES = {
        HandlerMode.CONTEXT:
            lambda self: f'Must match the "{self.input_argument}" field.'
    }

    def query(self):
        return self.value == self.argument

MaxLength

Bases: ConditionHandler[str | list | dict, int]

Ensures the collection or string does not exceed N items/characters

Source code in pipeline/handlers/condition_handler/condition.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class MaxLength(ConditionHandler[str | list | dict, int]):
    """Ensures the collection or string does not exceed N items/characters"""
    FLAGS = (ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR, )

    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Too long. Maximum length is {self.argument} characters."
            if isinstance(self.value, str) else
            f"Too many items. Maximum count is {self.argument}."
    }

    def query(self):
        return len(self.value) <= self.argument

MaxNumber

Bases: ConditionHandler[int | float, int | float]

Ensures the numeric value is less than or equal to N

Source code in pipeline/handlers/condition_handler/condition.py 
100
101
102
103
104
105
106
107
108
109
class MaxNumber(ConditionHandler[int | float, int | float]):
    """Ensures the numeric value is less than or equal to N"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda self: f"Value must be at most {self.argument}."
    }

    def query(self):
        return self.value <= self.argument

MinLength

Bases: ConditionHandler[str | list | dict, int]

Ensures the collection or string has at least N items/characters

Source code in pipeline/handlers/condition_handler/condition.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class MinLength(ConditionHandler[str | list | dict, int]):
    """Ensures the collection or string has at least N items/characters"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Too short. Minimum length is {self.argument} characters."
            if isinstance(self.value, str) else
            f"Too few items. Minimum count is {self.argument}."
    }

    def query(self):
        return len(self.value) >= self.argument

MinNumber

Bases: ConditionHandler[int | float, int | float]

Ensures the numeric value is greater than or equal to N

Source code in pipeline/handlers/condition_handler/condition.py 
88
89
90
91
92
93
94
95
96
97
98
class MinNumber(ConditionHandler[int | float, int | float]):
    """Ensures the numeric value is greater than or equal to N"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self: f"Value must be at least {self.argument}."
    }

    def query(self):
        return self.value >= self.argument

NotEqual

Bases: ConditionHandler[Any, Any]

Ensures the value is strictly not equal to the argument or a specific context field

Source code in pipeline/handlers/condition_handler/condition.py 
147
148
149
150
151
152
153
154
155
156
class NotEqual(ConditionHandler[Any, Any]):
    """Ensures the value is strictly not equal to the argument or a specific context field"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Value must be different."
    }

    def query(self):
        return self.value != self.argument

NotIncludedIn

Bases: ConditionHandler[Any, Iterable]

Ensures the value does not exist within the provided blacklist

Source code in pipeline/handlers/condition_handler/condition.py 
124
125
126
127
128
129
130
131
132
133
class NotIncludedIn(ConditionHandler[Any, Iterable]):
    """Ensures the value does not exist within the provided blacklist"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "This value is not allowed."
    }

    def query(self):
        return self.value not in self.argument

Pipeline

Bases: ConditionHandler[dict, Pipeline]

Validates a dictionary using the same rules as the normal pipeline, but for nested data.

Source code in pipeline/handlers/condition_handler/condition.py 
183
184
185
186
187
188
189
190
191
192
193
194
class Pipeline(ConditionHandler[dict, Pipeline]):
    """Validates a dictionary using the same rules as the normal pipeline, but for nested data."""
    SUPPORT = (HandlerMode.ROOT, )

    ERROR_BUILDER = lambda self: self.metadata['errors']

    ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: None}

    def query(self):
        self.metadata['errors'] = self.argument.run(data=self.value).errors

        return self.metadata['errors'] is None

ValueType

Bases: ConditionHandler[Any, type]

A built-in condition handler to validate the type of the value.

This handler is automatically used by the Pipe to ensure the passed value matches the expected type defined in the Pipe.

Source code in pipeline/handlers/condition_handler/condition.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class ValueType(ConditionHandler[Any, type]):
    """
    A built-in condition handler to validate the type of the value.

    This handler is automatically used by the Pipe to ensure the passed value matches
    the expected type defined in the Pipe.
    """
    FLAGS = (ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR, )

    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self: f"Invalid type. Expected {self.argument.__name__}."
    }

    def query(self):
        return isinstance(self.value, self.argument)

pipeline.handlers.match_handler.match_handler

MatchHandler

Bases: ConditionHandler[V, A]

Base class for match handlers.

Match handlers extend condition handlers to provide specific matching capabilities, often involving regular expressions or patterns.

Source code in pipeline/handlers/match_handler/match_handler.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class MatchHandler(ConditionHandler[V, A]):
    """
    Base class for match handlers.

    Match handlers extend condition handlers to provide specific matching capabilities,
    often involving regular expressions or patterns.
    """
    def search(
        self,
        pattern: str | re.Pattern,
        flag: Optional[re.RegexFlag] = None
    ) -> bool:
        """
        Searches for the pattern in the value.

        Args:
            pattern (str | re.Pattern): The regex pattern to search for.
            flag (Optional[re.RegexFlag]): Optional regex flags.

        Returns:
            bool: True if the pattern is found, False otherwise.
        """
        return re.search(pattern, str(self.value), flag or 0) is not None

    def fullmatch(
        self,
        pattern: str | re.Pattern,
        flag: Optional[re.RegexFlag] = None
    ) -> bool:
        """
        Checks if the entire value matches the pattern.

        Args:
            pattern (str | re.Pattern): The regex pattern to match against.
            flag (Optional[re.RegexFlag]): Optional regex flags.

        Returns:
            bool: True if the entire value matches the pattern, False otherwise.
        """
        return re.fullmatch(pattern, str(self.value), flag or 0) is not None

fullmatch(pattern, flag=None)

Checks if the entire value matches the pattern.

Parameters:

Name Type Description Default
pattern str | Pattern

The regex pattern to match against.

 required
flag Optional[RegexFlag]

Optional regex flags.

 None

Returns:

Name Type Description
bool bool

True if the entire value matches the pattern, False otherwise.
Source code in pipeline/handlers/match_handler/match_handler.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def fullmatch(
    self,
    pattern: str | re.Pattern,
    flag: Optional[re.RegexFlag] = None
) -> bool:
    """
    Checks if the entire value matches the pattern.

    Args:
        pattern (str | re.Pattern): The regex pattern to match against.
        flag (Optional[re.RegexFlag]): Optional regex flags.

    Returns:
        bool: True if the entire value matches the pattern, False otherwise.
    """
    return re.fullmatch(pattern, str(self.value), flag or 0) is not None

search(pattern, flag=None)

Searches for the pattern in the value.

Parameters:

Name Type Description Default
pattern str | Pattern

The regex pattern to search for.

 required
flag Optional[RegexFlag]

Optional regex flags.

 None

Returns:

Name Type Description
bool bool

True if the pattern is found, False otherwise.
Source code in pipeline/handlers/match_handler/match_handler.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def search(
    self,
    pattern: str | re.Pattern,
    flag: Optional[re.RegexFlag] = None
) -> bool:
    """
    Searches for the pattern in the value.

    Args:
        pattern (str | re.Pattern): The regex pattern to search for.
        flag (Optional[re.RegexFlag]): Optional regex flags.

    Returns:
        bool: True if the pattern is found, False otherwise.
    """
    return re.search(pattern, str(self.value), flag or 0) is not None

pipeline.handlers.match_handler.match

Match

Central registry for all match handler units.

This class provides a convenient way to access different match handlers (e.g., Text, Regex, Web) from a single location.

Source code in pipeline/handlers/match_handler/match.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class Match:
    """
    Central registry for all match handler units.

    This class provides a convenient way to access different match handlers
    (e.g., Text, Regex, Web) from a single location.
    """
    Text: ClassVar[Type[MatchText]] = MatchText
    Regex: ClassVar[Type[MatchRegex]] = MatchRegex

    Web: ClassVar[Type[MatchWeb]] = MatchWeb
    Network: ClassVar[Type[MatchNetwork]] = MatchNetwork

    Time: ClassVar[Type[MatchTime]] = MatchTime
    Localization: ClassVar[Type[MatchLocalization]] = MatchLocalization

    Format: ClassVar[Type[MatchFormat]] = MatchFormat
    Encoding: ClassVar[Type[MatchEncoding]] = MatchEncoding

MatchEncoding

Registry for encoding-related match handlers.

Includes handlers for Base64, JSON, etc.

Source code in pipeline/handlers/match_handler/units/match_encoding.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class MatchEncoding:
    """
    Registry for encoding-related match handlers.

    Includes handlers for Base64, JSON, etc.
    """
    class Base64(MatchHandler[str, None]):
        """Checks if string is valid Base64 encoded"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Invalid Base64 encoding."
        }

        def query(self):
            return self.fullmatch(
                r"^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$"
            )

    class JSON(MatchHandler[str, None]):
        """Validates that a string is a correctly formatted JSON object or array"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "String is not valid JSON."
        }

        def query(self):
            try:
                json.loads(self.value)

                return True
            except (ValueError, TypeError):
                return False

Base64

Bases: MatchHandler[str, None]

Checks if string is valid Base64 encoded

Source code in pipeline/handlers/match_handler/units/match_encoding.py 
15
16
17
18
19
20
21
22
23
24
25
26
class Base64(MatchHandler[str, None]):
    """Checks if string is valid Base64 encoded"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Invalid Base64 encoding."
    }

    def query(self):
        return self.fullmatch(
            r"^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$"
        )

JSON

Bases: MatchHandler[str, None]

Validates that a string is a correctly formatted JSON object or array

Source code in pipeline/handlers/match_handler/units/match_encoding.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class JSON(MatchHandler[str, None]):
    """Validates that a string is a correctly formatted JSON object or array"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "String is not valid JSON."
    }

    def query(self):
        try:
            json.loads(self.value)

            return True
        except (ValueError, TypeError):
            return False

MatchFormat

Registry for format-related match handlers.

Includes handlers for Email, UUID, HexColor, etc.

Source code in pipeline/handlers/match_handler/units/match_format.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class MatchFormat:
    """
    Registry for format-related match handlers.

    Includes handlers for Email, UUID, HexColor, etc.
    """
    class Email(MatchHandler[str, None]):
        """Accepts email addresses with standard user, domain, and TLD parts"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Invalid email address format (e.g., 'user@example.com')."
        }

        def query(self):
            return self.fullmatch(
                r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
            )

    class UUID(MatchHandler[str, None]):
        """Validates 36-character hexadecimal unique identifiers (8-4-4-4-12)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid UUID format."}

        def query(self):
            return self.fullmatch(
                r"^[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}$", re.IGNORECASE
            )

    class HexColor(MatchHandler[str, None]):
        """Accepts hex colors in 3 or 6 digit formats (e.g., #F00, #FF0000)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Must be a valid hex color code (e.g., #FFFFFF)."
        }

        def query(self):
            return self.fullmatch(r"^#(?:[0-9a-fA-F]{3}){1,2}$")

    class E164Phone(MatchHandler[str, None]):
        """International phone numbers in E.164 format (e.g., +1234567890)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Invalid phone number format. Must use international format (e.g., '+1234567890')."
        }

        def query(self):
            return self.fullmatch(r"^\+[1-9]\d{1,14}$")

    class Password(MatchHandler[str, str]):
        """Validates password strength based on three policies: RELAXED, NORMAL, or STRICT.

        Policies:
        - RELAXED: 6-64 chars, 1 uppercase, 1 lowercase.
        - NORMAL: 6-64 chars, 1 uppercase, 1 lowercase, 1 digit.
        - STRICT: 6-64 chars, 1 uppercase, 1 lowercase, 1 digit, 1 special character.

        Requires a policy argument (e.g., Match.Format.Password.NORMAL)
        """
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        RELAXED = "relaxed"
        NORMAL = "normal"
        STRICT = "strict"

        ERROR_MESSAGE = {
            RELAXED:
                "Password too weak. Required: 6-64 characters, at least 1 uppercase and 1 lowercase letter.",
            NORMAL:
                "Password too weak. Required: 6-64 characters, at least 1 uppercase, 1 lowercase, and 1 digit.",
            STRICT:
                "Password too weak. Required: 6-64 characters, at least 1 uppercase, 1 lowercase, 1 digit, and 1 special character."
        }

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self: self.ERROR_MESSAGE[self.argument]  # type: ignore
        }

        def query(self):
            if self.argument == self.RELAXED:
                # Min 6, Max 64, 1 Upper, 1 Lower
                pattern = r"^(?=.*[a-z])(?=.*[A-Z]).{6,64}$"
            elif self.argument == self.NORMAL:
                # Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit
                pattern = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).{6,64}$"
            elif self.argument == self.STRICT:
                # Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit, 1 Special
                pattern = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[!@#$%^&*(),.?\":{}|<>]).{6,64}$"
            else:
                raise HandlerException(
                    f"{self.argument} is not a valid password policy. Use Password.RELAXED, Password.NORMAL, or Password.STRICT."
                )

            return self.fullmatch(pattern)

    class JWT(MatchHandler[str, None]):
        """Validates the structure of a JSON Web Token (header.payload.signature)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid JWT format."}

        def query(self):
            return self.fullmatch(
                r"^[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+$"
            )

E164Phone

Bases: MatchHandler[str, None]

International phone numbers in E.164 format (e.g., +1234567890)

Source code in pipeline/handlers/match_handler/units/match_format.py 
55
56
57
58
59
60
61
62
63
64
65
66
class E164Phone(MatchHandler[str, None]):
    """International phone numbers in E.164 format (e.g., +1234567890)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Invalid phone number format. Must use international format (e.g., '+1234567890')."
    }

    def query(self):
        return self.fullmatch(r"^\+[1-9]\d{1,14}$")

Email

Bases: MatchHandler[str, None]

Accepts email addresses with standard user, domain, and TLD parts

Source code in pipeline/handlers/match_handler/units/match_format.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Email(MatchHandler[str, None]):
    """Accepts email addresses with standard user, domain, and TLD parts"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Invalid email address format (e.g., 'user@example.com')."
    }

    def query(self):
        return self.fullmatch(
            r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
        )

HexColor

Bases: MatchHandler[str, None]

Accepts hex colors in 3 or 6 digit formats (e.g., #F00, #FF0000)

Source code in pipeline/handlers/match_handler/units/match_format.py 
43
44
45
46
47
48
49
50
51
52
53
class HexColor(MatchHandler[str, None]):
    """Accepts hex colors in 3 or 6 digit formats (e.g., #F00, #FF0000)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Must be a valid hex color code (e.g., #FFFFFF)."
    }

    def query(self):
        return self.fullmatch(r"^#(?:[0-9a-fA-F]{3}){1,2}$")

JWT

Bases: MatchHandler[str, None]

Validates the structure of a JSON Web Token (header.payload.signature)

Source code in pipeline/handlers/match_handler/units/match_format.py 
115
116
117
118
119
120
121
122
123
124
class JWT(MatchHandler[str, None]):
    """Validates the structure of a JSON Web Token (header.payload.signature)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid JWT format."}

    def query(self):
        return self.fullmatch(
            r"^[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+\.[A-Za-z0-9-_]+$"
        )

Password

Bases: MatchHandler[str, str]

Validates password strength based on three policies: RELAXED, NORMAL, or STRICT.

Policies: - RELAXED: 6-64 chars, 1 uppercase, 1 lowercase. - NORMAL: 6-64 chars, 1 uppercase, 1 lowercase, 1 digit. - STRICT: 6-64 chars, 1 uppercase, 1 lowercase, 1 digit, 1 special character.

Requires a policy argument (e.g., Match.Format.Password.NORMAL)

Source code in pipeline/handlers/match_handler/units/match_format.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class Password(MatchHandler[str, str]):
    """Validates password strength based on three policies: RELAXED, NORMAL, or STRICT.

    Policies:
    - RELAXED: 6-64 chars, 1 uppercase, 1 lowercase.
    - NORMAL: 6-64 chars, 1 uppercase, 1 lowercase, 1 digit.
    - STRICT: 6-64 chars, 1 uppercase, 1 lowercase, 1 digit, 1 special character.

    Requires a policy argument (e.g., Match.Format.Password.NORMAL)
    """
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    RELAXED = "relaxed"
    NORMAL = "normal"
    STRICT = "strict"

    ERROR_MESSAGE = {
        RELAXED:
            "Password too weak. Required: 6-64 characters, at least 1 uppercase and 1 lowercase letter.",
        NORMAL:
            "Password too weak. Required: 6-64 characters, at least 1 uppercase, 1 lowercase, and 1 digit.",
        STRICT:
            "Password too weak. Required: 6-64 characters, at least 1 uppercase, 1 lowercase, 1 digit, and 1 special character."
    }

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self: self.ERROR_MESSAGE[self.argument]  # type: ignore
    }

    def query(self):
        if self.argument == self.RELAXED:
            # Min 6, Max 64, 1 Upper, 1 Lower
            pattern = r"^(?=.*[a-z])(?=.*[A-Z]).{6,64}$"
        elif self.argument == self.NORMAL:
            # Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit
            pattern = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d).{6,64}$"
        elif self.argument == self.STRICT:
            # Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit, 1 Special
            pattern = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[!@#$%^&*(),.?\":{}|<>]).{6,64}$"
        else:
            raise HandlerException(
                f"{self.argument} is not a valid password policy. Use Password.RELAXED, Password.NORMAL, or Password.STRICT."
            )

        return self.fullmatch(pattern)

UUID

Bases: MatchHandler[str, None]

Validates 36-character hexadecimal unique identifiers (8-4-4-4-12)

Source code in pipeline/handlers/match_handler/units/match_format.py 
32
33
34
35
36
37
38
39
40
41
class UUID(MatchHandler[str, None]):
    """Validates 36-character hexadecimal unique identifiers (8-4-4-4-12)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid UUID format."}

    def query(self):
        return self.fullmatch(
            r"^[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}$", re.IGNORECASE
        )

MatchLocalization

Registry for localization-related match handlers.

Includes handlers for Country, Currency, Language (ISO codes), and Timezones.

Source code in pipeline/handlers/match_handler/units/match_localization.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class MatchLocalization:
    """
    Registry for localization-related match handlers.

    Includes handlers for Country, Currency, Language (ISO codes), and Timezones.
    """
    class Country(MatchHandler[str, None]):
        """ISO 3166-1 alpha-2 (e.g., 'US', 'DE', 'JP')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must be a valid 2-letter ISO country code (e.g., US)."
        }

        def query(self):
            return self.value in ISO_3166

    class Currency(MatchHandler[str, None]):
        """ISO 4217 (e.g., 'USD', 'EUR', 'BTC')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Must be a valid 3-letter currency code (e.g., USD)."
        }

        def query(self):
            return self.value in ISO_4217

    class Language(MatchHandler[str, None]):
        """ISO 639-1 (e.g., 'en', 'fr', 'zh')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Must be a valid 2-letter language code (e.g., en)."
        }

        def query(self):
            return self.value in ISO_639_1

    class Timezone(MatchHandler[str, None]):
        """IANA Timezone (e.g., 'America/New_York', 'Europe/London')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Invalid IANA timezone string (e.g., America/New_York)."
        }

        def query(self):
            return self.value in available_timezones()

Country

Bases: MatchHandler[str, None]

ISO 3166-1 alpha-2 (e.g., 'US', 'DE', 'JP')

Source code in pipeline/handlers/match_handler/units/match_localization.py 
18
19
20
21
22
23
24
25
26
27
28
29
class Country(MatchHandler[str, None]):
    """ISO 3166-1 alpha-2 (e.g., 'US', 'DE', 'JP')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must be a valid 2-letter ISO country code (e.g., US)."
    }

    def query(self):
        return self.value in ISO_3166

Currency

Bases: MatchHandler[str, None]

ISO 4217 (e.g., 'USD', 'EUR', 'BTC')

Source code in pipeline/handlers/match_handler/units/match_localization.py 
31
32
33
34
35
36
37
38
39
40
41
class Currency(MatchHandler[str, None]):
    """ISO 4217 (e.g., 'USD', 'EUR', 'BTC')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Must be a valid 3-letter currency code (e.g., USD)."
    }

    def query(self):
        return self.value in ISO_4217

Language

Bases: MatchHandler[str, None]

ISO 639-1 (e.g., 'en', 'fr', 'zh')

Source code in pipeline/handlers/match_handler/units/match_localization.py 
43
44
45
46
47
48
49
50
51
52
53
class Language(MatchHandler[str, None]):
    """ISO 639-1 (e.g., 'en', 'fr', 'zh')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Must be a valid 2-letter language code (e.g., en)."
    }

    def query(self):
        return self.value in ISO_639_1

Timezone

Bases: MatchHandler[str, None]

IANA Timezone (e.g., 'America/New_York', 'Europe/London')

Source code in pipeline/handlers/match_handler/units/match_localization.py 
55
56
57
58
59
60
61
62
63
64
65
66
class Timezone(MatchHandler[str, None]):
    """IANA Timezone (e.g., 'America/New_York', 'Europe/London')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Invalid IANA timezone string (e.g., America/New_York)."
    }

    def query(self):
        return self.value in available_timezones()

MatchNetwork

Registry for network-related match handlers.

Includes handlers for IPv4, IPv6, MAC Addresses, etc.

Source code in pipeline/handlers/match_handler/units/match_network.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class MatchNetwork:
    """
    Registry for network-related match handlers.

    Includes handlers for IPv4, IPv6, MAC Addresses, etc.
    """
    class IPv4(MatchHandler[str, None]):
        """Validates a standard IPv4 address (e.g., '192.168.1.1')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid IPv4 address."}

        def query(self):
            try:
                return ip_address(address=self.value).version == 4
            except:
                return False

    class IPv6(MatchHandler[str, None]):
        """Validates an IPv6 address (e.g., '2001:db8::ff00:42:8329')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid IPv6 address."}

        def query(self):
            try:
                return ip_address(address=self.value).version == 6
            except:
                return False

    class MACAddress(MatchHandler[str, None]):
        """Accepts hardware MAC addresses using colon, hyphen, or dot separators"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Invalid MAC address format."
        }

        def query(self):
            return self.fullmatch(
                r"^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})|([0-9a-fA-F]{4}\.[0-9a-fA-F]{4}\.[0-9a-fA-F]{4})$"
            )

IPv4

Bases: MatchHandler[str, None]

Validates a standard IPv4 address (e.g., '192.168.1.1')

Source code in pipeline/handlers/match_handler/units/match_network.py 
15
16
17
18
19
20
21
22
23
24
25
class IPv4(MatchHandler[str, None]):
    """Validates a standard IPv4 address (e.g., '192.168.1.1')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid IPv4 address."}

    def query(self):
        try:
            return ip_address(address=self.value).version == 4
        except:
            return False

IPv6

Bases: MatchHandler[str, None]

Validates an IPv6 address (e.g., '2001:db8::ff00:42:8329')

Source code in pipeline/handlers/match_handler/units/match_network.py 
27
28
29
30
31
32
33
34
35
36
37
class IPv6(MatchHandler[str, None]):
    """Validates an IPv6 address (e.g., '2001:db8::ff00:42:8329')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {HandlerMode.ROOT: lambda _: "Invalid IPv6 address."}

    def query(self):
        try:
            return ip_address(address=self.value).version == 6
        except:
            return False

MACAddress

Bases: MatchHandler[str, None]

Accepts hardware MAC addresses using colon, hyphen, or dot separators

Source code in pipeline/handlers/match_handler/units/match_network.py 
39
40
41
42
43
44
45
46
47
48
49
50
class MACAddress(MatchHandler[str, None]):
    """Accepts hardware MAC addresses using colon, hyphen, or dot separators"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Invalid MAC address format."
    }

    def query(self):
        return self.fullmatch(
            r"^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})|([0-9a-fA-F]{4}\.[0-9a-fA-F]{4}\.[0-9a-fA-F]{4})$"
        )

MatchRegex

Registry for regex-related match handlers.

Includes handlers for Search and FullMatch.

Source code in pipeline/handlers/match_handler/units/match_regex.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class MatchRegex:
    """
    Registry for regex-related match handlers.

    Includes handlers for Search and FullMatch.
    """
    class Search(MatchHandler[str, str | Pattern]):
        """Accepts values that contain at least one match of the provided regex pattern"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Invalid value. Valid pattern for value is {self.argument}."
        }

        def query(self):
            return self.search(self.argument)

    class FullMatch(MatchHandler[str, str | Pattern]):
        """Accepts values that match the provided regex pattern in their entirety"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Invalid value. Valid pattern for value is {self.argument}."
        }

        def query(self):
            return self.fullmatch(self.argument)

FullMatch

Bases: MatchHandler[str, str | Pattern]

Accepts values that match the provided regex pattern in their entirety

Source code in pipeline/handlers/match_handler/units/match_regex.py 
28
29
30
31
32
33
34
35
36
37
38
39
class FullMatch(MatchHandler[str, str | Pattern]):
    """Accepts values that match the provided regex pattern in their entirety"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Invalid value. Valid pattern for value is {self.argument}."
    }

    def query(self):
        return self.fullmatch(self.argument)

Search

Bases: MatchHandler[str, str | Pattern]

Accepts values that contain at least one match of the provided regex pattern

Source code in pipeline/handlers/match_handler/units/match_regex.py 
15
16
17
18
19
20
21
22
23
24
25
26
class Search(MatchHandler[str, str | Pattern]):
    """Accepts values that contain at least one match of the provided regex pattern"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Invalid value. Valid pattern for value is {self.argument}."
    }

    def query(self):
        return self.search(self.argument)

MatchText

Registry for text-related match handlers.

Includes handlers for Lowercase, Uppercase, Digits, etc.

Source code in pipeline/handlers/match_handler/units/match_text.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class MatchText:
    """
    Registry for text-related match handlers.

    Includes handlers for Lowercase, Uppercase, Digits, etc.
    """
    class Lowercase(MatchHandler[str, None]):
        """Accepts ONLY lowercase English letters (a-z)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Must contain only lowercase letters (a-z)."
        }

        def query(self):
            return self.fullmatch(r"^[a-z]+$")

    class LowercaseWithSpaces(MatchHandler[str, None]):
        """Accepts lowercase English letters (a-z) and spaces"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only lowercase letters and spaces (e.g., 'hello world')."
        }

        def query(self):
            return self.fullmatch(r"^[a-z ]+$")

    class Uppercase(MatchHandler[str, None]):
        """Accepts ONLY uppercase English letters (A-Z)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Must contain only uppercase letters (A-Z)."
        }

        def query(self):
            return self.fullmatch(r"^[A-Z]+$")

    class UppercaseWithSpaces(MatchHandler[str, None]):
        """Accepts uppercase English letters (A-Z) and spaces"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only uppercase letters and spaces (e.g., 'HELLO WORLD')."
        }

        def query(self):
            return self.fullmatch(r"^[A-Z ]+$")

    class Letters(MatchHandler[str, None]):
        """Accepts ONLY case English letters (a-z, A-Z)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Must contain only letters (a-z, A-Z)."
        }

        def query(self):
            return self.fullmatch(r"^[a-zA-Z]+$")

    class LettersWithSpaces(MatchHandler[str, None]):
        """Accepts English letters (a-z, A-Z) and spaces"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only letters and spaces (e.g., 'Hello World')."
        }

        def query(self):
            return self.fullmatch(r"^[a-zA-Z ]+$")

    class Digits(MatchHandler[str, None]):
        """Accepts ONLY numeric digits (0-9)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Must contain only digits (0-9)."
        }

        def query(self):
            return self.fullmatch(r"^\d+$")

    class DigitsWithSpaces(MatchHandler[str, None]):
        """Accepts numeric digits (0-9) and spaces"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only digits and spaces (e.g., '123 456')."
        }

        def query(self):
            return self.fullmatch(r"^[\d ]+$")

    class Alphanumeric(MatchHandler[str, None]):
        """Accepts letters and numeric digits. No symbols or spaces"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Must contain only letters and digits (e.g., 'abc123')."
        }

        def query(self):
            return self.fullmatch(r"^[a-zA-Z0-9]+$")

    class AlphanumericWithSpaces(MatchHandler[str, None]):
        """Accepts letters, numeric digits, and spaces. No symbols"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only letters, digits, and spaces (e.g., 'abc 123')."
        }

        def query(self):
            return self.fullmatch(r"^[a-zA-Z0-9 ]+$")

    class Printable(MatchHandler[str, None]):
        """Accepts letters, numbers, symbols, and spaces (ASCII 20-7E)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Contains invalid characters. Only printable ASCII characters are allowed."
        }

        def query(self):
            return self.fullmatch(r"^[ -~]+$")

    class NoWhitespace(MatchHandler[str, None]):
        """Ensures string contains no spaces, tabs, or line breaks"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Must not contain spaces, tabs, or line breaks."
        }

        def query(self):
            return not self.search(r"\s")

    class Slug(MatchHandler[str, None]):
        """URL-friendly strings: 'my-cool-post-123'"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only lowercase letters, numbers, and hyphens (e.g., 'my-post-123')."
        }

        def query(self):
            return self.fullmatch(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")

Alphanumeric

Bases: MatchHandler[str, None]

Accepts letters and numeric digits. No symbols or spaces

Source code in pipeline/handlers/match_handler/units/match_text.py 
109
110
111
112
113
114
115
116
117
118
119
class Alphanumeric(MatchHandler[str, None]):
    """Accepts letters and numeric digits. No symbols or spaces"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Must contain only letters and digits (e.g., 'abc123')."
    }

    def query(self):
        return self.fullmatch(r"^[a-zA-Z0-9]+$")

AlphanumericWithSpaces

Bases: MatchHandler[str, None]

Accepts letters, numeric digits, and spaces. No symbols

Source code in pipeline/handlers/match_handler/units/match_text.py 
121
122
123
124
125
126
127
128
129
130
131
132
class AlphanumericWithSpaces(MatchHandler[str, None]):
    """Accepts letters, numeric digits, and spaces. No symbols"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only letters, digits, and spaces (e.g., 'abc 123')."
    }

    def query(self):
        return self.fullmatch(r"^[a-zA-Z0-9 ]+$")

Digits

Bases: MatchHandler[str, None]

Accepts ONLY numeric digits (0-9)

Source code in pipeline/handlers/match_handler/units/match_text.py 
85
86
87
88
89
90
91
92
93
94
class Digits(MatchHandler[str, None]):
    """Accepts ONLY numeric digits (0-9)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Must contain only digits (0-9)."
    }

    def query(self):
        return self.fullmatch(r"^\d+$")

DigitsWithSpaces

Bases: MatchHandler[str, None]

Accepts numeric digits (0-9) and spaces

Source code in pipeline/handlers/match_handler/units/match_text.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class DigitsWithSpaces(MatchHandler[str, None]):
    """Accepts numeric digits (0-9) and spaces"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only digits and spaces (e.g., '123 456')."
    }

    def query(self):
        return self.fullmatch(r"^[\d ]+$")

Letters

Bases: MatchHandler[str, None]

Accepts ONLY case English letters (a-z, A-Z)

Source code in pipeline/handlers/match_handler/units/match_text.py 
61
62
63
64
65
66
67
68
69
70
class Letters(MatchHandler[str, None]):
    """Accepts ONLY case English letters (a-z, A-Z)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Must contain only letters (a-z, A-Z)."
    }

    def query(self):
        return self.fullmatch(r"^[a-zA-Z]+$")

LettersWithSpaces

Bases: MatchHandler[str, None]

Accepts English letters (a-z, A-Z) and spaces

Source code in pipeline/handlers/match_handler/units/match_text.py 
72
73
74
75
76
77
78
79
80
81
82
83
class LettersWithSpaces(MatchHandler[str, None]):
    """Accepts English letters (a-z, A-Z) and spaces"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only letters and spaces (e.g., 'Hello World')."
    }

    def query(self):
        return self.fullmatch(r"^[a-zA-Z ]+$")

Lowercase

Bases: MatchHandler[str, None]

Accepts ONLY lowercase English letters (a-z)

Source code in pipeline/handlers/match_handler/units/match_text.py 
13
14
15
16
17
18
19
20
21
22
class Lowercase(MatchHandler[str, None]):
    """Accepts ONLY lowercase English letters (a-z)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Must contain only lowercase letters (a-z)."
    }

    def query(self):
        return self.fullmatch(r"^[a-z]+$")

LowercaseWithSpaces

Bases: MatchHandler[str, None]

Accepts lowercase English letters (a-z) and spaces

Source code in pipeline/handlers/match_handler/units/match_text.py 
24
25
26
27
28
29
30
31
32
33
34
35
class LowercaseWithSpaces(MatchHandler[str, None]):
    """Accepts lowercase English letters (a-z) and spaces"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only lowercase letters and spaces (e.g., 'hello world')."
    }

    def query(self):
        return self.fullmatch(r"^[a-z ]+$")

NoWhitespace

Bases: MatchHandler[str, None]

Ensures string contains no spaces, tabs, or line breaks

Source code in pipeline/handlers/match_handler/units/match_text.py 
147
148
149
150
151
152
153
154
155
156
157
class NoWhitespace(MatchHandler[str, None]):
    """Ensures string contains no spaces, tabs, or line breaks"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Must not contain spaces, tabs, or line breaks."
    }

    def query(self):
        return not self.search(r"\s")

Printable

Bases: MatchHandler[str, None]

Accepts letters, numbers, symbols, and spaces (ASCII 20-7E)

Source code in pipeline/handlers/match_handler/units/match_text.py 
134
135
136
137
138
139
140
141
142
143
144
145
class Printable(MatchHandler[str, None]):
    """Accepts letters, numbers, symbols, and spaces (ASCII 20-7E)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Contains invalid characters. Only printable ASCII characters are allowed."
    }

    def query(self):
        return self.fullmatch(r"^[ -~]+$")

Slug

Bases: MatchHandler[str, None]

URL-friendly strings: 'my-cool-post-123'

Source code in pipeline/handlers/match_handler/units/match_text.py 
159
160
161
162
163
164
165
166
167
168
169
170
class Slug(MatchHandler[str, None]):
    """URL-friendly strings: 'my-cool-post-123'"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only lowercase letters, numbers, and hyphens (e.g., 'my-post-123')."
    }

    def query(self):
        return self.fullmatch(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")

Uppercase

Bases: MatchHandler[str, None]

Accepts ONLY uppercase English letters (A-Z)

Source code in pipeline/handlers/match_handler/units/match_text.py 
37
38
39
40
41
42
43
44
45
46
class Uppercase(MatchHandler[str, None]):
    """Accepts ONLY uppercase English letters (A-Z)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Must contain only uppercase letters (A-Z)."
    }

    def query(self):
        return self.fullmatch(r"^[A-Z]+$")

UppercaseWithSpaces

Bases: MatchHandler[str, None]

Accepts uppercase English letters (A-Z) and spaces

Source code in pipeline/handlers/match_handler/units/match_text.py 
48
49
50
51
52
53
54
55
56
57
58
59
class UppercaseWithSpaces(MatchHandler[str, None]):
    """Accepts uppercase English letters (A-Z) and spaces"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only uppercase letters and spaces (e.g., 'HELLO WORLD')."
    }

    def query(self):
        return self.fullmatch(r"^[A-Z ]+$")

MatchTime

Registry for time-related match handlers.

Includes handlers for Date, Time, and DateTime (ISO 8601).

Source code in pipeline/handlers/match_handler/units/match_time.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class MatchTime:
    """
    Registry for time-related match handlers.

    Includes handlers for Date, Time, and DateTime (ISO 8601).
    """
    class Date(MatchHandler[str, None]):
        """Validates YYYY-MM-DD format"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Date must be in YYYY-MM-DD format (e.g., 2023-10-01)."
        }

        def query(self):
            return self.fullmatch(
                r"^\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$"
            )

    class Time(MatchHandler[str, None]):
        """Validates 24h time in HH:MM or HH:MM:SS format"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _: "Invalid time format (HH:MM[:SS]) (e.g., 14:30:00)."
        }

        def query(self):
            return self.fullmatch(r"^(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d)?$")

    class DateTime(MatchHandler[str, None]):
        """Validates ISO 8601 combined Date and Time (e.g., 2023-10-01T14:30:00Z)."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Invalid ISO 8601 DateTime format (e.g., 2023-10-01T14:30:00Z)."
        }

        def query(self):
            return self.fullmatch(
                r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})$"
            )

Date

Bases: MatchHandler[str, None]

Validates YYYY-MM-DD format

Source code in pipeline/handlers/match_handler/units/match_time.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class Date(MatchHandler[str, None]):
    """Validates YYYY-MM-DD format"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Date must be in YYYY-MM-DD format (e.g., 2023-10-01)."
    }

    def query(self):
        return self.fullmatch(
            r"^\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$"
        )

DateTime

Bases: MatchHandler[str, None]

Validates ISO 8601 combined Date and Time (e.g., 2023-10-01T14:30:00Z).

Source code in pipeline/handlers/match_handler/units/match_time.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class DateTime(MatchHandler[str, None]):
    """Validates ISO 8601 combined Date and Time (e.g., 2023-10-01T14:30:00Z)."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Invalid ISO 8601 DateTime format (e.g., 2023-10-01T14:30:00Z)."
    }

    def query(self):
        return self.fullmatch(
            r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})$"
        )

Time

Bases: MatchHandler[str, None]

Validates 24h time in HH:MM or HH:MM:SS format

Source code in pipeline/handlers/match_handler/units/match_time.py 
28
29
30
31
32
33
34
35
36
37
38
class Time(MatchHandler[str, None]):
    """Validates 24h time in HH:MM or HH:MM:SS format"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Invalid time format (HH:MM[:SS]) (e.g., 14:30:00)."
    }

    def query(self):
        return self.fullmatch(r"^(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d)?$")

MatchWeb

Registry for web-related match handlers.

Includes handlers for Domain and URL.

Source code in pipeline/handlers/match_handler/units/match_web.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class MatchWeb:
    """
    Registry for web-related match handlers.

    Includes handlers for Domain and URL.
    """

    class Domain(MatchHandler[str, None]):
        """Validates a domain name based on RFC 1035."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda _: "Invalid domain format (e.g., 'example.com')."
        }

        def query(self):
            return self.fullmatch(
                r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$"
            )

    class URL(MatchHandler[str, None]):
        """Validates web URLs using HTTP or HTTPS protocols."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Invalid URL format. Must be a valid HTTP/HTTPS URL (e.g., 'https://example.com')."
        }

        def query(self):
            return self.fullmatch(
                r"^https?://(?:www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b(?:[-a-zA-Z0-9()@:%_\+.~#?&//=]*)$"
            )

Domain

Bases: MatchHandler[str, None]

Validates a domain name based on RFC 1035.

Source code in pipeline/handlers/match_handler/units/match_web.py 
14
15
16
17
18
19
20
21
22
23
24
25
class Domain(MatchHandler[str, None]):
    """Validates a domain name based on RFC 1035."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Invalid domain format (e.g., 'example.com')."
    }

    def query(self):
        return self.fullmatch(
            r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$"
        )

URL

Bases: MatchHandler[str, None]

Validates web URLs using HTTP or HTTPS protocols.

Source code in pipeline/handlers/match_handler/units/match_web.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class URL(MatchHandler[str, None]):
    """Validates web URLs using HTTP or HTTPS protocols."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Invalid URL format. Must be a valid HTTP/HTTPS URL (e.g., 'https://example.com')."
    }

    def query(self):
        return self.fullmatch(
            r"^https?://(?:www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b(?:[-a-zA-Z0-9()@:%_\+.~#?&//=]*)$"
        )

pipeline.handlers.transform_handler.transform_handler

TransformHandler

Bases: BaseHandler[V, A]

Abstract base class for transform handlers.

Transform handlers modify the input value and return the transformed value. They must implement the operation method.

Source code in pipeline/handlers/transform_handler/transform_handler.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class TransformHandler(BaseHandler[V, A]):
    """
    Abstract base class for transform handlers.

    Transform handlers modify the input value and return the transformed value.
    They must implement the `operation` method.
    """
    @abstractmethod
    def operation(self) -> V:
        """
        Performs the transformation operation.

        Returns:
            V: The transformed value.
        """
        ...

    def _handle(self) -> V:
        """
        Handles the transformation in ROOT or CONTEXT mode.

        Returns:
            V: The transformed value.
        """
        return self.operation()

    def _handle_item_mode(self) -> V:
        """
        Handles the transformation in ITEM mode (for iterables).

        Iterates over the input value and applies the transformation to each item.
        The input value is modified in place if it's a list or dict.

        Returns:
            V: The transformed iterable (same object as input, but modified).

        Raises:
            HandlerModeException: If the input value is not a list or dict.
        """
        if isinstance(self.input_value, list):
            items = enumerate(self.input_value)
        elif isinstance(self.input_value, dict):
            items = self.input_value.items()
        else:
            raise HandlerModeException(
                "The transform handler input value is not of type list or dict."
            )

        for key, value in items:
            if self._item_use_key:
                value = key

            if not self._is_valid_type(value, self._expected_value_type):
                continue

            # NOTE: We use can cast() here because we checked if the value type is valid but linter does not know that.
            self.value = cast(V, value)

            self.input_value[key] = self.operation()

        return self.input_value

operation() abstractmethod

Performs the transformation operation.

Returns:

Name Type Description
V V

The transformed value.
Source code in pipeline/handlers/transform_handler/transform_handler.py 
18
19
20
21
22
23
24
25
26
@abstractmethod
def operation(self) -> V:
    """
    Performs the transformation operation.

    Returns:
        V: The transformed value.
    """
    ...

pipeline.handlers.transform_handler.transform

Transform

Central registry for all transform handlers.

Transform handlers modify the value in some way, such as changing case, replacing substrings, or performing arithmetic operations.

Source code in pipeline/handlers/transform_handler/transform.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class Transform:
    """
    Central registry for all transform handlers.

    Transform handlers modify the value in some way, such as changing case,
    replacing substrings, or performing arithmetic operations.
    """
    class Strip(TransformHandler[str, Optional[str]]):
        """Removes leading and trailing whitespace from a string"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.value.strip()

    class Capitalize(TransformHandler[str, None]):
        """Converts the first character to uppercase and the rest to lowercase"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.value.capitalize()

    class Lowercase(TransformHandler[str, None]):
        """Converts all characters in the string to lowercase"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.value.lower()

    class Uppercase(TransformHandler[str, None]):
        """Converts all characters in the string to uppercase"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.value.upper()

    class Multiply(TransformHandler[str | list | int | float, int | float]):
        """
        Multiplies a string, list, integer or float by the provided argument

        If the value is not numeric, the argument is rounded before multiplication.
        """
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            if not isinstance(self.value, (int, float)):
                return self.value * round(self.argument)

            return self.value * self.argument

    class Reverse(TransformHandler[str | list, None]):
        """Reverses the order of characters in a string or items in a list"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.value[::-1]

    class Title(TransformHandler[str, None]):
        """Converts the first character of every word to uppercase"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.value.title()

    class SnakeCase(TransformHandler[str, None]):
        """Converts strings to snake_case (e.g., 'HelloWorld' -> 'hello_world')"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            s = re.sub(r'(?<!^)(?=[A-Z])', '_', self.value).lower()

            return s.replace("-", "_").replace(" ", "_")

    class Unique(TransformHandler[list, None]):
        """Removes duplicate items from a list while preserving order"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return list(dict.fromkeys(self.value))

    class Replace(TransformHandler[str, tuple]):
        """Replaces all occurrences of a substring with another (old, new)"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            old, new = self.argument

            return self.value.replace(old, new)

    class Apply(TransformHandler[Any, Callable]):
        """
        Transforms the value by passing it through the provided callable.

        The callable should take only one argument (the value) and return the 
        transformed value. No checks are performed on the returned value, 
        so ensure it meets your specific requirements.
        """
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        def operation(self):
            return self.argument(self.value)

Apply

Bases: TransformHandler[Any, Callable]

Transforms the value by passing it through the provided callable.

The callable should take only one argument (the value) and return the transformed value. No checks are performed on the returned value, so ensure it meets your specific requirements.

Source code in pipeline/handlers/transform_handler/transform.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
class Apply(TransformHandler[Any, Callable]):
    """
    Transforms the value by passing it through the provided callable.

    The callable should take only one argument (the value) and return the 
    transformed value. No checks are performed on the returned value, 
    so ensure it meets your specific requirements.
    """
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.argument(self.value)

Capitalize

Bases: TransformHandler[str, None]

Converts the first character to uppercase and the rest to lowercase

Source code in pipeline/handlers/transform_handler/transform.py 
23
24
25
26
27
28
class Capitalize(TransformHandler[str, None]):
    """Converts the first character to uppercase and the rest to lowercase"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.value.capitalize()

Lowercase

Bases: TransformHandler[str, None]

Converts all characters in the string to lowercase

Source code in pipeline/handlers/transform_handler/transform.py 
30
31
32
33
34
35
class Lowercase(TransformHandler[str, None]):
    """Converts all characters in the string to lowercase"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.value.lower()

Multiply

Bases: TransformHandler[str | list | int | float, int | float]

Multiplies a string, list, integer or float by the provided argument

If the value is not numeric, the argument is rounded before multiplication.

Source code in pipeline/handlers/transform_handler/transform.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
class Multiply(TransformHandler[str | list | int | float, int | float]):
    """
    Multiplies a string, list, integer or float by the provided argument

    If the value is not numeric, the argument is rounded before multiplication.
    """
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        if not isinstance(self.value, (int, float)):
            return self.value * round(self.argument)

        return self.value * self.argument

Replace

Bases: TransformHandler[str, tuple]

Replaces all occurrences of a substring with another (old, new)

Source code in pipeline/handlers/transform_handler/transform.py 
88
89
90
91
92
93
94
95
class Replace(TransformHandler[str, tuple]):
    """Replaces all occurrences of a substring with another (old, new)"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        old, new = self.argument

        return self.value.replace(old, new)

Reverse

Bases: TransformHandler[str | list, None]

Reverses the order of characters in a string or items in a list

Source code in pipeline/handlers/transform_handler/transform.py 
58
59
60
61
62
63
class Reverse(TransformHandler[str | list, None]):
    """Reverses the order of characters in a string or items in a list"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.value[::-1]

SnakeCase

Bases: TransformHandler[str, None]

Converts strings to snake_case (e.g., 'HelloWorld' -> 'hello_world')

Source code in pipeline/handlers/transform_handler/transform.py 
72
73
74
75
76
77
78
79
class SnakeCase(TransformHandler[str, None]):
    """Converts strings to snake_case (e.g., 'HelloWorld' -> 'hello_world')"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        s = re.sub(r'(?<!^)(?=[A-Z])', '_', self.value).lower()

        return s.replace("-", "_").replace(" ", "_")

Strip

Bases: TransformHandler[str, Optional[str]]

Removes leading and trailing whitespace from a string

Source code in pipeline/handlers/transform_handler/transform.py 
16
17
18
19
20
21
class Strip(TransformHandler[str, Optional[str]]):
    """Removes leading and trailing whitespace from a string"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.value.strip()

Title

Bases: TransformHandler[str, None]

Converts the first character of every word to uppercase

Source code in pipeline/handlers/transform_handler/transform.py 
65
66
67
68
69
70
class Title(TransformHandler[str, None]):
    """Converts the first character of every word to uppercase"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.value.title()

Unique

Bases: TransformHandler[list, None]

Removes duplicate items from a list while preserving order

Source code in pipeline/handlers/transform_handler/transform.py 
81
82
83
84
85
86
class Unique(TransformHandler[list, None]):
    """Removes duplicate items from a list while preserving order"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return list(dict.fromkeys(self.value))

Uppercase

Bases: TransformHandler[str, None]

Converts all characters in the string to uppercase

Source code in pipeline/handlers/transform_handler/transform.py 
37
38
39
40
41
42
class Uppercase(TransformHandler[str, None]):
    """Converts all characters in the string to uppercase"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    def operation(self):
        return self.value.upper()

pipeline.integration.falcon.decorator

process_request(pre_hook=None, post_hook=None, handle_errors=None, **pipes_config)

A decorator factory that validates Falcon request (WSGI and ASGI) data using a defined pipeline.

This decorator extracts data from a Falcon Request object (either from query parameters for GET requests or the media body for other methods), runs it through a PipelineFalcon instance, and injects the validated data into the decorated responder method.

If validation fails, it automatically sets the response body to the validation errors and returns a 422 Unprocessable Content status, preventing the responder from executing. You can use your own error handler via PipelineFalcon.handle_errors, but it must end the request.

Parameters:

Name Type Description Default
pre_hook PipelineHookFunc | None

A function to be called before each pipe execution. The global_pre_hook will not run if a local pre_hook is defined.

 None
post_hook PipelineHookFunc | None

A function to be called after each pipe execution. The global_post_hook will not run if a local pre_hook is defined.

 None
handle_errors PipelineHandleErrorsFunc | None

A function to handle errors collected during pipeline execution. This could be used to raise exceptions, log errors, or format them for a response. The global_handle_errors will not run if a local handle_errors is defined.

 None
**pipes_config PipelinePipeConfig

Configuration for the pipes. Keys represent the fields in the data dictionary to be processed, and values are the configuration for the corresponding pipe.

 {}

Returns:

Name Type Description
Callable

A decorator that wraps a Falcon responder method.
Example

@request_validator( email={ "type": str, "conditions": { Pipe.Condition.MaxLength: 64 }, "matches": { Pipe.Match.Format.Email: None } } ) def on_post(self, req, resp, email): pass

Source code in pipeline/integration/falcon/decorator.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def process_request(
    pre_hook: PipelineHookFunc | None = None,
    post_hook: PipelineHookFunc | None = None,
    handle_errors: PipelineHandleErrorsFunc | None = None,
    **pipes_config: PipelinePipeConfig
):
    """
    A decorator factory that validates Falcon request (WSGI and ASGI) data
    using a defined pipeline.

    This decorator extracts data from a Falcon `Request` object (either from 
    query parameters for GET requests or the media body for other methods), 
    runs it through a `PipelineFalcon` instance, and injects the validated 
    data into the decorated responder method.

    If validation fails, it automatically sets the response body to the 
    validation errors and returns a 422 Unprocessable Content status, preventing 
    the responder from executing. You can use your own error handler via
    `PipelineFalcon.handle_errors`, but it must end the request.

    Args:
        pre_hook (PipelineHookFunc | None): A function to be called before each pipe execution.
            The global_pre_hook will not run if a local pre_hook is defined.
        post_hook (PipelineHookFunc | None): A function to be called after each pipe execution.
            The global_post_hook will not run if a local pre_hook is defined.
        handle_errors (PipelineHandleErrorsFunc | None): A function to handle
            errors collected during pipeline execution. This could be used to raise exceptions,
            log errors, or format them for a response. The global_handle_errors will not run
            if a local handle_errors is defined.
        **pipes_config (PipelinePipeConfig): Configuration for the pipes.
            Keys represent the fields in the data dictionary to be processed,
            and values are the configuration for the corresponding pipe.

    Returns:
        Callable: A decorator that wraps a Falcon responder method.

    Example:
        @request_validator(
            email={
                "type": str,
                "conditions": {
                    Pipe.Condition.MaxLength: 64
                },
                "matches": {
                    Pipe.Match.Format.Email: None
                }
            }
        )
        def on_post(self, req, resp, email):
            pass
    """
    def decorator(responder: Callable):
        is_async: bool = inspect.iscoroutinefunction(responder)

        def get_params(req: RequestSync | RequestAsync) -> dict:
            data: dict = cast(dict, req.params)

            for field, value in data.items():
                try:
                    data[field] = json.loads(value)
                except (json.JSONDecodeError, TypeError):
                    pass

            return data

        def run_pipeline(data: dict) -> PipelineResult:
            return PipelineFalcon(
                pre_hook=pre_hook,
                post_hook=post_hook,
                handle_errors=handle_errors,
                **pipes_config
            ).run(data=data)

        def handle_result(
            resp: ResponseSync | ResponseAsync, result: PipelineResult
        ) -> bool:
            if result.errors:
                resp.media = result.errors
                resp.status = 422

                return True

            return False

        if is_async:

            @wraps(responder)
            async def async_wrapper(
                resource: object, req: RequestAsync, resp: ResponseAsync, *args,
                **kwargs
            ):
                data: dict

                if req.method.upper() == "GET":
                    data = get_params(req=req)
                else:
                    data = await req.get_media({})

                result: PipelineResult = run_pipeline(data)

                if handle_result(
                    resp=resp, result=result
                ) or result.processed_data is None:
                    return

                return await responder(
                    resource, req, resp, *args, **kwargs,
                    **result.processed_data
                )

            return async_wrapper
        else:

            @wraps(responder)
            def sync_wrapper(
                resource: object, req: RequestSync, resp: ResponseAsync, *args,
                **kwargs
            ):
                data: dict

                if req.method.upper() == "GET":
                    data = get_params(req=req)
                else:
                    data = req.get_media({})

                result: PipelineResult = run_pipeline(data)

                if handle_result(
                    resp=resp, result=result
                ) or result.processed_data is None:
                    return

                return responder(
                    resource, req, resp, *args, **kwargs,
                    **result.processed_data
                )

            return sync_wrapper

    return decorator