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. 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. 4. Matches: Only if no condition errors occurred, match handlers are executed. These are typically regex-based checks. 5. 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

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. 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.
    4. Matches: Only if no condition errors occurred, match handlers are executed. These are typically
       regex-based checks.
    5. 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,
        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`).
            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.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[V]:
        """
        Executes the pipe processing logic.

        The method strictly follows the defined order of operations. 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.optional and (bool(self.value) is False):
            return PipeResult(
                value=self.value, condition_errors=[], match_errors=[]
            )

        if (error := self.Condition.ValueType(self.value, self.type).handle()):
            return PipeResult(
                value=self.value, condition_errors=[error], match_errors=[]
            )

        if self.conditions:
            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

        if len(self._condition_errors) == 0:
            if self.matches:
                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

            if self.transform and len(self._match_errors) == 0:
                for handler, argument in self.transform.items():
                    self.value = handler(
                        value=self.value,
                        argument=argument,
                        context=self.context
                    ).handle()

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

__init__(value, type, 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
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

def __init__(
    self,
    value: V,
    type: T,
    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`).
        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.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 strictly follows the defined order of operations. 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[V]	PipeResult[V]: The result containing the processed value (or original value if errors occurred)
PipeResult[V]	and lists of any condition or match errors.

Source code in pipeline/core/pipe/pipe.py

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

    The method strictly follows the defined order of operations. 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.optional and (bool(self.value) is False):
        return PipeResult(
            value=self.value, condition_errors=[], match_errors=[]
        )

    if (error := self.Condition.ValueType(self.value, self.type).handle()):
        return PipeResult(
            value=self.value, condition_errors=[error], match_errors=[]
        )

    if self.conditions:
        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

    if len(self._condition_errors) == 0:
        if self.matches:
            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

        if self.transform and len(self._match_errors) == 0:
            for handler, argument in self.transform.items():
                self.value = handler(
                    value=self.value,
                    argument=argument,
                    context=self.context
                ).handle()

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

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_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

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_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_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,
        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.
            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.pipes_config: dict[str, PipelinePipeConfig] = pipes_config

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

    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.

        After processing all fields, 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.
        """
        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._errors and self.handle_errors:
            self.handle_errors(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 Pipeline.global_pre_hook:
            Pipeline.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 Pipeline.global_post_hook:
            Pipeline.global_post_hook(hook)

        self._processed_data[field] = value

    def __call__(self, func: Any) -> Any:
        """
        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, **kwargs):
            result: PipelineResult = self.run(data=kwargs)

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

            kwargs = {**kwargs, **result.processed_data}

            return func(*args, **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.

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

def __call__(self, func: Any) -> Any:
    """
    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, **kwargs):
        result: PipelineResult = self.run(data=kwargs)

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

        kwargs = {**kwargs, **result.processed_data}

        return func(*args, **kwargs)

    return wrapper

__init__(pre_hook=None, post_hook=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
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

def __init__(
    self,
    pre_hook: PipelineHookFunc | None = None,
    post_hook: PipelineHookFunc | 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.
        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.pipes_config: dict[str, PipelinePipeConfig] = pipes_config

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

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.

After processing all fields, 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

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.

    After processing all fields, 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.
    """
    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._errors and self.handle_errors:
        self.handle_errors(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

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

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

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

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

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

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(
                "The condition/match handler input value is not of type 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

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

@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

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 format. Please provide a {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. This must be at least {self.argument} characters."
                if isinstance(self.value, str) else
                f"Please select at least {self.argument} items."
        }

        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"You can select a maximum of {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"Must be {self.argument} or greater."
        }

        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"Must be {self.argument} or less."
        }

        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 self:
                f"This value is not allowed. Please use a different value that is not in {list(self.argument)}."
        }

        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 self: f"The value must match {self.argument}."
        }

        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 self: f"Value cannot be equal to {self.argument}."
        }

        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"This 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"This cannot be the same as 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."""
        FLAGS = (ConditionFlag.RETURN_ONLY_ERROR_MSG, )

        SUPPORT = (HandlerMode.ROOT, )

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda self: self.metadata['errors']
        }

        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

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"This cannot be the same as 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

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 self: f"The value must match {self.argument}."
    }

    def query(self):
        return 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

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

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"This 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

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"You can select a maximum of {self.argument} items."
    }

    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

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"Must be {self.argument} or less."
    }

    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

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. This must be at least {self.argument} characters."
            if isinstance(self.value, str) else
            f"Please select at least {self.argument} items."
    }

    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

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"Must be {self.argument} or greater."
    }

    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

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 self: f"Value cannot be equal to {self.argument}."
    }

    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

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 self:
            f"This value is not allowed. Please use a different value that is not in {list(self.argument)}."
    }

    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

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

    SUPPORT = (HandlerMode.ROOT, )

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda self: self.metadata['errors']
    }

    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

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 format. Please provide a {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

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

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

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

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

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

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

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

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."
        }

        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 international phone format (E.164)."
        }

        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: "Min 6, Max 64, 1 Upper, 1 Lower",
            NORMAL: "Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit",
            STRICT: "Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit, 1 Special"
        }

        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)

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

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 international phone format (E.164)."
    }

    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

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."
    }

    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

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}$")

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

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: "Min 6, Max 64, 1 Upper, 1 Lower",
        NORMAL: "Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit",
        STRICT: "Min 6, Max 64, 1 Upper, 1 Lower, 1 Digit, 1 Special"
    }

    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

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

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."
        }

        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."
        }

        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."
        }

        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."
        }

        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

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."
    }

    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

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."
    }

    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

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."
    }

    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

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."
    }

    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

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

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

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

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

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

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

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

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 _: "Can only contain lowercase letters."
        }

        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 _: "Can only contain uppercase letters."
        }

        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 _: "Can only contain letters."
        }

        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 _: "Can only contain digits."
        }

        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 _: "Can only contain letters and digits."
        }

        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 or non-printable characters."
        }

        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 _: "Cannot contain spaces or tabs."
        }

        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 be lowercase letters, numbers, and hyphens."
        }

        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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Can only contain letters and digits."
    }

    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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Can only contain digits."
    }

    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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Can only contain letters."
    }

    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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Can only contain lowercase letters."
    }

    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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Cannot contain spaces or tabs."
    }

    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

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 or non-printable characters."
    }

    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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _: "Must be lowercase letters, numbers, and hyphens."
    }

    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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda _: "Can only contain uppercase letters."
    }

    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

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."
        }

        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])."
        }

        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."
        }

        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

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."
    }

    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

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."
    }

    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

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])."
    }

    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

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."}

        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."}

        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

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."}

    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

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."}

    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

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

@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

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)

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

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

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

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

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

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

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

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

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

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

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 400 Bad Request 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

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 400 Bad Request 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 = 400

                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