Technical Reference

handlers.base.cls

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[Callable | None]	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/cls.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[Callable | None]): 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[Callable[['BaseHandler', Any], Any] |
                                       None] = None

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

        Args:
            value (V): The value to process.
            argument (A): The argument for the handler.
            context (PipeContext | None): Additional context for the handler.
            metadata (PipeMetadata | None): Metadata about the pipe execution.
            _mode (HandlerMode): The mode in which the handler is operating.
            _item_use_key (bool | None): If True and in ITEM mode, the handler operates on the
                key of a dictionary item instead of the value.
            _preferred_value_type (type | None): 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: int | str | None = None
        self._item_use_key: bool | None = _item_use_key

        self._preferred_value_type: type | None = _preferred_value_type

        self._prepare_and_validate_handler()

    def __init_subclass__(cls) -> None:
        """
        Extracts expected runtime types from class generics and sets a unique handler ID.

        During subclass creation, this method:
        1. Extracts generic type arguments for `value` and `argument` from class definition,
           recursively unpacking them into tuples stored in `cls._raw_expected_type` for runtime type validation.
        2. Generates a `snake_case` identifier (`cls.id`) from the class name.
        """
        super().__init_subclass__()

        def _extract_types(arg: Any, is_top: bool = True) -> Any:
            args: tuple[Any, ...] = get_args(arg)

            if not args:
                return (arg, ) if is_top else arg

            return tuple(_extract_types(x, is_top=False) for x in args)

        value_type, argument_type = map(
            _extract_types, get_args(get_original_bases(cls)[0])
        )

        cls._raw_expected_type = HandlerExpectedType(
            value=value_type, argument=argument_type
        )

        partial_snake: str = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', cls.__name__)

        cls.id = re.sub('([a-z0-9])([A-Z])', r'\1_\2', partial_snake).lower()

    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: tuple[type, ...]
    ) -> bool:
        """
        Checks if a value matches the expected type(s).

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

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

        return 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_type.value):
                raise HandlerInvalidValueType(handler=self)

        if not self._is_valid_type(self.argument, self._expected_type.argument):
            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.
        """
        ...

    @cached_property
    def _expected_type(self) -> HandlerExpectedType:
        """
        Resolves the final target runtime types for validating `self.value` and `self.argument`.

        Returns the default `_raw_expected_type` parsed during subclass creation. If an instance-level
        `_preferred_value_type` was provided, it verifies compatibility against the allowed generic
        types and narrows down the expected `value` type to the preferred one.

        Returns:
            HandlerExpectedType: Container holding the expected type tuple(s) for `value` and `argument`.

        Raises:
            HandlerInvalidPreferredValueType: If `_preferred_value_type` is not compatible with the handler.
        """
        if not self._preferred_value_type:
            return self._raw_expected_type

        if self._preferred_value_type not in self._raw_expected_type.value and Any not in self._raw_expected_type.value:
            raise HandlerInvalidPreferredValueType(self)

        return HandlerExpectedType(
            value=(self._preferred_value_type, ),
            argument=self._raw_expected_type.argument
        )

__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	PipeContext | None	Additional context for the handler.	None
metadata	PipeMetadata | None	Metadata about the pipe execution.	None
_mode	HandlerMode	The mode in which the handler is operating.	ROOT
_item_use_key	bool | None	If True and in ITEM mode, the handler operates on the key of a dictionary item instead of the value.	False
_preferred_value_type	type | None	Specific type to prefer/enforce during type validation.	None

Source code in pipeline/handlers/base/cls.py

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

    Args:
        value (V): The value to process.
        argument (A): The argument for the handler.
        context (PipeContext | None): Additional context for the handler.
        metadata (PipeMetadata | None): Metadata about the pipe execution.
        _mode (HandlerMode): The mode in which the handler is operating.
        _item_use_key (bool | None): If True and in ITEM mode, the handler operates on the
            key of a dictionary item instead of the value.
        _preferred_value_type (type | None): 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: int | str | None = None
    self._item_use_key: bool | None = _item_use_key

    self._preferred_value_type: type | None = _preferred_value_type

    self._prepare_and_validate_handler()

__init_subclass__()

Extracts expected runtime types from class generics and sets a unique handler ID.

During subclass creation, this method: 1. Extracts generic type arguments for value and argument from class definition, recursively unpacking them into tuples stored in cls._raw_expected_type for runtime type validation. 2. Generates a snake_case identifier (cls.id) from the class name.

Source code in pipeline/handlers/base/cls.py

def __init_subclass__(cls) -> None:
    """
    Extracts expected runtime types from class generics and sets a unique handler ID.

    During subclass creation, this method:
    1. Extracts generic type arguments for `value` and `argument` from class definition,
       recursively unpacking them into tuples stored in `cls._raw_expected_type` for runtime type validation.
    2. Generates a `snake_case` identifier (`cls.id`) from the class name.
    """
    super().__init_subclass__()

    def _extract_types(arg: Any, is_top: bool = True) -> Any:
        args: tuple[Any, ...] = get_args(arg)

        if not args:
            return (arg, ) if is_top else arg

        return tuple(_extract_types(x, is_top=False) for x in args)

    value_type, argument_type = map(
        _extract_types, get_args(get_original_bases(cls)[0])
    )

    cls._raw_expected_type = HandlerExpectedType(
        value=value_type, argument=argument_type
    )

    partial_snake: str = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', cls.__name__)

    cls.id = re.sub('([a-z0-9])([A-Z])', r'\1_\2', partial_snake).lower()

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/cls.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.")

handlers.base.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/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	bool | None	If True, the handler uses the item's key (e.g., in a dictionary) instead of value.	False
only_consider	type | None	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/modifiers.py

def Item(
    handler: type[T] | partial[T],
    use_key: bool | None = False,
    only_consider: type | None = 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 (bool | None): If True, the handler uses the item's key (e.g., in a dictionary) instead of value.
        only_consider (type | None): 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
    )

handlers.condition.cls

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/cls.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: PipeContext | None = None,
        metadata: PipeMetadata | None = None,
        _mode: HandlerMode = HandlerMode.ROOT,
        _item_use_key: bool | None = False,
        _preferred_value_type: type | None = 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) -> ConditionError | None:
        """
        Handles the condition check in ROOT or CONTEXT mode.

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

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

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

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

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

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

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

            if not self._is_valid_type(value, self._expected_type.value):
                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/cls.py

def __init__(
    self,
    value: V,
    argument: A,
    context: PipeContext | None = None,
    metadata: PipeMetadata | None = None,
    _mode: HandlerMode = HandlerMode.ROOT,
    _item_use_key: bool | None = False,
    _preferred_value_type: type | None = 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/cls.py

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

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

handlers.condition.registry

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/registry.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 type. Expected {self.argument.__name__}."
        }

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

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

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

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

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

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

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

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

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            return self.metadata['errors'] is None

DoesNotMatchField

Bases: ConditionHandler[Any, Any]

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

Source code in pipeline/handlers/condition/registry.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'Must be different from the "{self.input_argument}" field.'
    }

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

Equal

Bases: ConditionHandler[Any, Any]

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

Source code in pipeline/handlers/condition/registry.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 _: "Value does not match the expected value."
    }

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

ExactLength

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

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

Source code in pipeline/handlers/condition/registry.py

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

    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

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

IncludedIn

Bases: ConditionHandler[Any, Iterable]

Ensures the value exists within the provided Iterable

Source code in pipeline/handlers/condition/registry.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/registry.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'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/registry.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"Too many items. Maximum count is {self.argument}."
    }

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

MaxNumber

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

Ensures the numeric value is less than or equal to N

Source code in pipeline/handlers/condition/registry.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"Value must be at most {self.argument}."
    }

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

MinLength

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

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

Source code in pipeline/handlers/condition/registry.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. Minimum length is {self.argument} characters."
            if isinstance(self.value, str) else
            f"Too few items. Minimum count is {self.argument}."
    }

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

MinNumber

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

Ensures the numeric value is greater than or equal to N

Source code in pipeline/handlers/condition/registry.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"Value must be at least {self.argument}."
    }

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

NotEqual

Bases: ConditionHandler[Any, Any]

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

Source code in pipeline/handlers/condition/registry.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 _: "Value must be different."
    }

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

NotIncludedIn

Bases: ConditionHandler[Any, Iterable]

Ensures the value does not exist within the provided blacklist

Source code in pipeline/handlers/condition/registry.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 _: "This value is not allowed."
    }

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

Pipeline

Bases: ConditionHandler[dict, Pipeline]

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

Source code in pipeline/handlers/condition/registry.py

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

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

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

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

        return self.metadata['errors'] is None

ValueType

Bases: ConditionHandler[Any, type]

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

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

Source code in pipeline/handlers/condition/registry.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 type. Expected {self.argument.__name__}."
    }

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

handlers.match.cls

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/cls.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: re.RegexFlag | None = None
    ) -> bool:
        """
        Searches for the pattern in the value.

        Args:
            pattern (str | re.Pattern): The regex pattern to search for.
            flag (re.RegexFlag | None): 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: re.RegexFlag | None = None
    ) -> bool:
        """
        Checks if the entire value matches the pattern.

        Args:
            pattern (str | re.Pattern): The regex pattern to match against.
            flag (re.RegexFlag | None): 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

    @staticmethod
    def get_diacritics(
        languages: tuple[str, ...] | None,
        letter_case: Literal["lower", "upper"] | None = None,
        /,
    ) -> str:
        """
        Retrieves a string of diacritic characters for the specified languages.

        Combines all unique diacritic marks associated with a given set of ISO 639-1
        language codes. The output can be filtered by letter case or return both
        cases by default.

        Args:
            languages: A tuple of ISO 639-1 language codes (e.g., ("fr", "de")). 
                If None or empty, an empty string is returned.
            letter_case: The desired grammatical case for the diacritics. 
                Options are "lower", "upper", or None. If None, both cases 
                are returned. This is a positional-only argument.

        Returns:
            A string containing the concatenated diacritic characters for the 
            requested languages and case.
        """
        if not languages:
            return ""

        def transform(base: str) -> str:
            if letter_case == "lower":
                return base

            if letter_case == "upper":
                return base.upper()

            return base + base.upper()

        return "".join(transform(ISO_639_1[language]) for language in languages)

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	RegexFlag | None	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/cls.py

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

    Args:
        pattern (str | re.Pattern): The regex pattern to match against.
        flag (re.RegexFlag | None): 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

get_diacritics(languages, letter_case=None) staticmethod

Retrieves a string of diacritic characters for the specified languages.

Combines all unique diacritic marks associated with a given set of ISO 639-1 language codes. The output can be filtered by letter case or return both cases by default.

Parameters:

Name	Type	Description	Default
languages	tuple[str, ...] | None	A tuple of ISO 639-1 language codes (e.g., ("fr", "de")). If None or empty, an empty string is returned.	required
letter_case	Literal['lower', 'upper'] | None	The desired grammatical case for the diacritics. Options are "lower", "upper", or None. If None, both cases are returned. This is a positional-only argument.	None

Returns:

Type	Description
str	A string containing the concatenated diacritic characters for the
str	requested languages and case.

Source code in pipeline/handlers/match/cls.py

@staticmethod
def get_diacritics(
    languages: tuple[str, ...] | None,
    letter_case: Literal["lower", "upper"] | None = None,
    /,
) -> str:
    """
    Retrieves a string of diacritic characters for the specified languages.

    Combines all unique diacritic marks associated with a given set of ISO 639-1
    language codes. The output can be filtered by letter case or return both
    cases by default.

    Args:
        languages: A tuple of ISO 639-1 language codes (e.g., ("fr", "de")). 
            If None or empty, an empty string is returned.
        letter_case: The desired grammatical case for the diacritics. 
            Options are "lower", "upper", or None. If None, both cases 
            are returned. This is a positional-only argument.

    Returns:
        A string containing the concatenated diacritic characters for the 
        requested languages and case.
    """
    if not languages:
        return ""

    def transform(base: str) -> str:
        if letter_case == "lower":
            return base

        if letter_case == "upper":
            return base.upper()

        return base + base.upper()

    return "".join(transform(ISO_639_1[language]) for language in languages)

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	RegexFlag | None	Optional regex flags.	None

Returns:

Name	Type	Description
bool	bool	True if the pattern is found, False otherwise.

Source code in pipeline/handlers/match/cls.py

def search(
    self,
    pattern: str | re.Pattern,
    flag: re.RegexFlag | None = None
) -> bool:
    """
    Searches for the pattern in the value.

    Args:
        pattern (str | re.Pattern): The regex pattern to search for.
        flag (re.RegexFlag | None): 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

handlers.match.registry

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/registry.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 = MatchText
    Regex: ClassVar = MatchRegex

    Web: ClassVar = MatchWeb
    Network: ClassVar = MatchNetwork

    Time: ClassVar = MatchTime
    Localization: ClassVar = MatchLocalization

    Format: ClassVar = MatchFormat
    Encoding: ClassVar = MatchEncoding

MatchEncoding

Registry for encoding-related match handlers.

Includes handlers for Base64, JSON, etc.

Source code in pipeline/handlers/match/units/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/units/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/units/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/units/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 (e.g., 'user@example.com')."
        }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

            return self.fullmatch(pattern)

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

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

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

E164Phone

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 phone number format. Must use international format (e.g., '+1234567890')."
    }

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

Email

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 (e.g., 'user@example.com')."
    }

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

HexColor

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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}$")

JWT

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/format.py

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

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

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

Password

Bases: MatchHandler[str, str]

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

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

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

Source code in pipeline/handlers/match/units/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:
            "Password too weak. Required: 6-64 characters, at least 1 uppercase and 1 lowercase letter.",
        NORMAL:
            "Password too weak. Required: 6-64 characters, at least 1 uppercase, 1 lowercase, and 1 digit.",
        STRICT:
            "Password too weak. Required: 6-64 characters, at least 1 uppercase, 1 lowercase, 1 digit, and 1 special character."
    }

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

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

        return self.fullmatch(pattern)

UUID

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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/units/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 (e.g., US)."
        }

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

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

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

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

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

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

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

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

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

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

Country

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 (e.g., US)."
    }

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

Currency

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 (e.g., USD)."
    }

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

Language

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 (e.g., en)."
    }

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

Timezone

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 (e.g., America/New_York)."
    }

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

MatchNetwork

Registry for network-related match handlers.

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

Source code in pipeline/handlers/match/units/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/units/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/units/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/units/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/units/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/units/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/units/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/units/text.py

class MatchText:
    """
    Registry for text-related match handlers.

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

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

        def query(self):
            diacritics: str = self.get_diacritics(self.argument, "lower")

            return self.fullmatch(rf"^[a-z{diacritics}]+$")

    class LowercaseWithSpaces(MatchHandler[str, tuple | None]):
        """Accepts lowercase English letters (a-z) and spaces if no diacritics are provided via tuple."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only lowercase letters, spaces and allowed diacritics."
        }

        def query(self):
            diacritics: str = self.get_diacritics(self.argument, "lower")

            return self.fullmatch(rf"^[a-z {diacritics}]+$")

    class Uppercase(MatchHandler[str, tuple | None]):
        """Accepts ONLY uppercase English letters (A-Z) if no diacritics are provided via tuple."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

        def query(self):
            diacritics: str = self.get_diacritics(self.argument, "upper")

            return self.fullmatch(rf"^[A-Z{diacritics}]+$")

    class UppercaseWithSpaces(MatchHandler[str, tuple | None]):
        """Accepts uppercase English letters (A-Z) and spaces if no diacritics are provided via tuple."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only uppercase letters, spaces and allowed diacritics."
        }

        def query(self):
            diacritics: str = self.get_diacritics(self.argument, "upper")

            return self.fullmatch(rf"^[A-Z {diacritics}]+$")

    class Letters(MatchHandler[str, tuple | None]):
        """Accepts ONLY English letters (a-z, A-Z) if no diacritics are provided via tuple."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

        def query(self):
            diacritics: str = self.get_diacritics(self.argument)

            return self.fullmatch(rf"^[a-zA-Z{diacritics}]+$")

    class LettersWithSpaces(MatchHandler[str, tuple | None]):
        """Accepts English letters (a-z, A-Z) and spaces if no diacritics are provided via tuple."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only letters, spaces and allowed diacritics."
        }

        def query(self):
            diacritics: str = self.get_diacritics(self.argument)

            return self.fullmatch(rf"^[a-zA-Z {diacritics}]+$")

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

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

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

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

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

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

    class Alphanumeric(MatchHandler[str, tuple | None]):
        """Accepts letters and digits if no diacritics are provided via tuple. No symbols or spaces."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only letters, digits and allowed diacritics."
        }

        def query(self):
            diacritics: str = self.get_diacritics(self.argument)

            return self.fullmatch(rf"^[a-zA-Z0-9{diacritics}]+$")

    class AlphanumericWithSpaces(MatchHandler[str, tuple | None]):
        """Accepts letters, digits, and spaces if no diacritics are provided via tuple. No symbols."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda _:
                "Must contain only letters, digits, spaces and allowed diacritics."
        }

        def query(self):
            diacritics: str = self.get_diacritics(self.argument)

            return self.fullmatch(rf"^[a-zA-Z0-9 {diacritics}]+$")

    class Printable(MatchHandler[str, tuple | None]):
        """Accepts ASCII (20-7E) and diacritics if provided via tuple."""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

        def query(self):
            diacritics: str = self.get_diacritics(self.argument)

            return self.fullmatch(rf"^[ -~{diacritics}]+$")

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

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

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

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

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

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

Alphanumeric

Bases: MatchHandler[str, tuple | None]

Accepts letters and digits if no diacritics are provided via tuple. No symbols or spaces.

Source code in pipeline/handlers/match/units/text.py

class Alphanumeric(MatchHandler[str, tuple | None]):
    """Accepts letters and digits if no diacritics are provided via tuple. No symbols or spaces."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only letters, digits and allowed diacritics."
    }

    def query(self):
        diacritics: str = self.get_diacritics(self.argument)

        return self.fullmatch(rf"^[a-zA-Z0-9{diacritics}]+$")

AlphanumericWithSpaces

Bases: MatchHandler[str, tuple | None]

Accepts letters, digits, and spaces if no diacritics are provided via tuple. No symbols.

Source code in pipeline/handlers/match/units/text.py

class AlphanumericWithSpaces(MatchHandler[str, tuple | None]):
    """Accepts letters, digits, and spaces if no diacritics are provided via tuple. No symbols."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only letters, digits, spaces and allowed diacritics."
    }

    def query(self):
        diacritics: str = self.get_diacritics(self.argument)

        return self.fullmatch(rf"^[a-zA-Z0-9 {diacritics}]+$")

Digits

Bases: MatchHandler[str, None]

Accepts ONLY numeric digits (0-9)

Source code in pipeline/handlers/match/units/text.py

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

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

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

DigitsWithSpaces

Bases: MatchHandler[str, None]

Accepts numeric digits (0-9) and spaces

Source code in pipeline/handlers/match/units/text.py

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

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

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

Letters

Bases: MatchHandler[str, tuple | None]

Accepts ONLY English letters (a-z, A-Z) if no diacritics are provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class Letters(MatchHandler[str, tuple | None]):
    """Accepts ONLY English letters (a-z, A-Z) if no diacritics are provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

    def query(self):
        diacritics: str = self.get_diacritics(self.argument)

        return self.fullmatch(rf"^[a-zA-Z{diacritics}]+$")

LettersWithSpaces

Bases: MatchHandler[str, tuple | None]

Accepts English letters (a-z, A-Z) and spaces if no diacritics are provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class LettersWithSpaces(MatchHandler[str, tuple | None]):
    """Accepts English letters (a-z, A-Z) and spaces if no diacritics are provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only letters, spaces and allowed diacritics."
    }

    def query(self):
        diacritics: str = self.get_diacritics(self.argument)

        return self.fullmatch(rf"^[a-zA-Z {diacritics}]+$")

Lowercase

Bases: MatchHandler[str, tuple | None]

Accepts ONLY lowercase English letters (a-z) if no diacritics are provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class Lowercase(MatchHandler[str, tuple | None]):
    """Accepts ONLY lowercase English letters (a-z) if no diacritics are provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

    def query(self):
        diacritics: str = self.get_diacritics(self.argument, "lower")

        return self.fullmatch(rf"^[a-z{diacritics}]+$")

LowercaseWithSpaces

Bases: MatchHandler[str, tuple | None]

Accepts lowercase English letters (a-z) and spaces if no diacritics are provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class LowercaseWithSpaces(MatchHandler[str, tuple | None]):
    """Accepts lowercase English letters (a-z) and spaces if no diacritics are provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only lowercase letters, spaces and allowed diacritics."
    }

    def query(self):
        diacritics: str = self.get_diacritics(self.argument, "lower")

        return self.fullmatch(rf"^[a-z {diacritics}]+$")

NoWhitespace

Bases: MatchHandler[str, None]

Ensures string contains no spaces, tabs, or line breaks

Source code in pipeline/handlers/match/units/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 _: "Must not contain spaces, tabs, or line breaks."
    }

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

Printable

Bases: MatchHandler[str, tuple | None]

Accepts ASCII (20-7E) and diacritics if provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class Printable(MatchHandler[str, tuple | None]):
    """Accepts ASCII (20-7E) and diacritics if provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

    def query(self):
        diacritics: str = self.get_diacritics(self.argument)

        return self.fullmatch(rf"^[ -~{diacritics}]+$")

Slug

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 contain only lowercase letters, numbers, and hyphens (e.g., 'my-post-123')."
    }

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

Uppercase

Bases: MatchHandler[str, tuple | None]

Accepts ONLY uppercase English letters (A-Z) if no diacritics are provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class Uppercase(MatchHandler[str, tuple | None]):
    """Accepts ONLY uppercase English letters (A-Z) if no diacritics are provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

    def query(self):
        diacritics: str = self.get_diacritics(self.argument, "upper")

        return self.fullmatch(rf"^[A-Z{diacritics}]+$")

UppercaseWithSpaces

Bases: MatchHandler[str, tuple | None]

Accepts uppercase English letters (A-Z) and spaces if no diacritics are provided via tuple.

Source code in pipeline/handlers/match/units/text.py

class UppercaseWithSpaces(MatchHandler[str, tuple | None]):
    """Accepts uppercase English letters (A-Z) and spaces if no diacritics are provided via tuple."""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda _:
            "Must contain only uppercase letters, spaces and allowed diacritics."
    }

    def query(self):
        diacritics: str = self.get_diacritics(self.argument, "upper")

        return self.fullmatch(rf"^[A-Z {diacritics}]+$")

MatchTime

Registry for time-related match handlers.

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

Source code in pipeline/handlers/match/units/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 (e.g., 2023-10-01)."
        }

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

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

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

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

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

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

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

Date

Bases: MatchHandler[str, None]

Validates YYYY-MM-DD format

Source code in pipeline/handlers/match/units/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 (e.g., 2023-10-01)."
    }

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

DateTime

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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 (e.g., 2023-10-01T14:30:00Z)."
    }

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

Time

Bases: MatchHandler[str, None]

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

Source code in pipeline/handlers/match/units/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]) (e.g., 14:30:00)."
    }

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

MatchWeb

Registry for web-related match handlers.

Includes handlers for Domain and URL.

Source code in pipeline/handlers/match/units/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 (e.g., 'example.com')."
        }

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

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

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

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

Domain

Bases: MatchHandler[str, None]

Validates a domain name based on RFC 1035.

Source code in pipeline/handlers/match/units/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 (e.g., 'example.com')."
    }

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

URL

Bases: MatchHandler[str, None]

Validates web URLs using HTTP or HTTPS protocols.

Source code in pipeline/handlers/match/units/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. Must be a valid HTTP/HTTPS URL (e.g., 'https://example.com')."
    }

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

handlers.transform.cls

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/cls.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_type.value):
                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/cls.py

@abstractmethod
def operation(self) -> V:
    """
    Performs the transformation operation.

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

handlers.transform.registry

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/registry.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, str | None]):
        """Removes leading and trailing whitespace from a string"""
        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

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

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

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

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

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

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

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

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

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

            return self.value * self.argument

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

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

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

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

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

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

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

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

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

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

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

            return self.value.replace(old, new)

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

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

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

Apply

Bases: TransformHandler[Any, Callable]

Transforms the value by passing it through the provided callable.

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

Source code in pipeline/handlers/transform/registry.py

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

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

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

Capitalize

Bases: TransformHandler[str, None]

Converts the first character to uppercase and the rest to lowercase

Source code in pipeline/handlers/transform/registry.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/registry.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/registry.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/registry.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/registry.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/registry.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, str | None]

Removes leading and trailing whitespace from a string

Source code in pipeline/handlers/transform/registry.py

class Strip(TransformHandler[str, str | None]):
    """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/registry.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/registry.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/registry.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()

integration.falcon.decorator

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

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

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

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

Parameters:

Name	Type	Description	Default
pre_hook	PipelineHookFunc | None	A function to be called before each pipe execution. The global_pre_hook will not run if a local pre_hook is defined.	None
post_hook	PipelineHookFunc | None	A function to be called after each pipe execution. The global_post_hook will not run if a local pre_hook is defined.	None
teardown	PipelineTeardownFunc | None	A function that will run after the entire pipeline finishes execution. This runs before handle_errors and will execute even if the pipeline failed. The global_teardown will not run if a local teardown is defined.	None
handle_errors	PipelineHandleErrorsFunc | None	A function to handle errors collected during pipeline execution. This could be used to raise exceptions, log errors, or format them for a response. The global_handle_errors will not run if a local handle_errors is defined.	None
**pipes_config	PipeConfig	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": { Condition.MaxLength: 64 }, "matches": { 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,
    teardown: PipelineTeardownFunc | None = None,
    handle_errors: PipelineHandleErrorsFunc | None = None,
    **pipes_config: PipeConfig
):
    """
    A decorator factory that validates Falcon request (WSGI and ASGI) data
    using a defined pipeline.

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

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

    Args:
        pre_hook (PipelineHookFunc | None): A function to be called before each pipe execution.
            The global_pre_hook will not run if a local pre_hook is defined.
        post_hook (PipelineHookFunc | None): A function to be called after each pipe execution.
            The global_post_hook will not run if a local pre_hook is defined.
        teardown (PipelineTeardownFunc | None): A function that will run after the entire 
            pipeline finishes execution. This runs before handle_errors and will execute 
            even if the pipeline failed. The global_teardown will not run if a local 
            teardown is defined.
        handle_errors (PipelineHandleErrorsFunc | None): A function to handle
            errors collected during pipeline execution. This could be used to raise exceptions,
            log errors, or format them for a response. The global_handle_errors will not run
            if a local handle_errors is defined.
        **pipes_config (PipeConfig): 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": {
                    Condition.MaxLength: 64
                },
                "matches": {
                    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,
                teardown=teardown,
                handle_errors=handle_errors,
                **pipes_config
            ).run(data=data)

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

                return True

            return False

        if is_async:

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

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

                result: PipelineResult = run_pipeline(data)

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

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

            return async_wrapper
        else:

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

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

                result: PipelineResult = run_pipeline(data)

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

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

            return sync_wrapper

    return decorator

battery.cls

Battery

A centralized registry of reusable pipe configurations for DRY validation.

Battery serves as a collection of pre-built BatteryUnit instances that represent common validation patterns (e.g., UUID, email, pagination limits). These units can be registered statically as class attributes or added dynamically at runtime, and then consumed anywhere in the application via the sink() method.

Source code in pipeline/battery/cls.py

class Battery:
    """
    A centralized registry of reusable pipe configurations for DRY validation.

    Battery serves as a collection of pre-built `BatteryUnit` instances that
    represent common validation patterns (e.g., UUID, email, pagination limits).
    These units can be registered statically as class attributes or added
    dynamically at runtime, and then consumed anywhere in the application
    via the `sink()` method.
    """
    UUID = BatteryUnit(
        type=str,
        conditions={Condition.ExactLength: 36},
        matches={Match.Format.UUID: None}
    )
    """A pre-built unit that validates a UUID v4 string."""
    @classmethod
    def add(cls, name: str, **pipe_config: Unpack[PipeConfig]) -> None:
        """
        Registers a new BatteryUnit as a class attribute.

        Creates a `BatteryUnit` from the provided pipe configuration and
        assigns it to the class under the given name. This allows dynamically
        extending the Battery registry at application startup.

        Args:
            name (str): The attribute name under which the unit will be stored on the class.
            **pipe_config (PipeConfig): The pipe configuration for the new unit.
        """
        setattr(cls, name, BatteryUnit(**pipe_config))

    @classmethod
    def get(cls, name: str) -> BatteryUnit:
        """
        Retrieves a registered BatteryUnit by name.

        Args:
            name (str): The attribute name of the unit to retrieve.

        Returns:
            BatteryUnit: The BatteryUnit instance registered under the given name.

        Raises:
            AttributeError: If no unit with the given name has been registered on the class.
        """
        return getattr(cls, name)

    @staticmethod
    def use_email(min_length: int = 6, max_length: int = 64) -> BatteryUnit:
        """
        Creates a BatteryUnit for email address validation.

        Produces a unit that validates a string value as a well-formed email
        address within a configurable length range.

        Args:
            min_length (int): The minimum allowed length of the email address.
                Defaults to 6.
            max_length (int): The maximum allowed length of the email address.
                Defaults to 64.

        Returns:
            BatteryUnit: A unit for email validation.
        """
        return BatteryUnit(
            type=str,
            conditions={
                Condition.MinLength: min_length,
                Condition.MaxLength: max_length
            },
            matches={Match.Format.Email: None},
            transform={Transform.Lowercase: None}
        )

    @staticmethod
    def use_limit(min_number: int = 1, max_number: int = 1000) -> BatteryUnit:
        """
        Creates a BatteryUnit for pagination limit validation.

        Produces a unit that validates an integer value representing the
        number of items to return per page, within a configurable range.

        Args:
            min_number (int): The minimum allowed limit value. Defaults to 1.
            max_number (int): The maximum allowed limit value. Defaults to 1000.

        Returns:
            BatteryUnit: A unit for pagination limit validation.
        """
        return BatteryUnit(
            type=int,
            conditions={
                Condition.MinNumber: min_number,
                Condition.MaxNumber: max_number
            }
        )

    @staticmethod
    def use_offset(min_number: int = 0, max_number: int = 1000) -> BatteryUnit:
        """
        Creates a BatteryUnit for pagination offset validation.

        Produces a unit that validates an integer value representing the
        starting position of a result set, within a configurable range.
        Differs from `use_limit` only in its default `min_number` of 0,
        which allows an offset of zero as a valid starting point.

        Args:
            min_number (int): The minimum allowed offset value. Defaults to 0.
            max_number (int): The maximum allowed offset value. Defaults to 1000.

        Returns:
            BatteryUnit: A unit for pagination offset validation.
        """
        return BatteryUnit(
            type=int,
            conditions={
                Condition.MinNumber: min_number,
                Condition.MaxNumber: max_number
            }
        )

    @staticmethod
    def use_token(
        exact_length: int = 32, digits_only: bool = False
    ) -> BatteryUnit:
        """
        Creates a BatteryUnit for token validation.

        Produces a unit that validates a string as a fixed-length token.
        Can be configured for alphanumeric characters or strictly digits.

        Args:
            exact_length (int): The required exact length of the token. 
                Defaults to 32.
            digits_only (bool): If True, validates using Match.Text.Digits.
                If False, validates using Match.Text.Alphanumeric. Defaults to False.

        Returns:
            BatteryUnit: A unit for token validation.
        """
        pattern = Match.Text.Digits if digits_only else Match.Text.Alphanumeric

        return BatteryUnit(
            type=str,
            conditions={Condition.ExactLength: exact_length},
            matches={pattern: None}
        )

    @staticmethod
    def use_password(
        policy: Literal["relaxed", "normal", "strict"] | None
    ) -> BatteryUnit:
        """
        Creates a BatteryUnit for password validation.

        Produces a unit that validates a string's length and, optionally, 
        its complexity against predefined security patterns.

        Args:
            policy (Literal["relaxed", "normal", "strict"]): The complexity level. If None, only
                length is validated (min: 6, max: 64).

        Returns:
            BatteryUnit: A unit for password validation.
        """
        config: PipeConfig = {
            'type': str,
            'conditions': {
                Condition.MinLength: 6,
                Condition.MaxLength: 64
            }
        }

        if policy is not None:
            config['matches'] = {Match.Format.Password: policy}

        return BatteryUnit(**config)

UUID = BatteryUnit(type=str, conditions={Condition.ExactLength: 36}, matches={Match.Format.UUID: None}) class-attribute instance-attribute

A pre-built unit that validates a UUID v4 string.

add(name, **pipe_config) classmethod

Registers a new BatteryUnit as a class attribute.

Creates a BatteryUnit from the provided pipe configuration and assigns it to the class under the given name. This allows dynamically extending the Battery registry at application startup.

Parameters:

Name	Type	Description	Default
name	str	The attribute name under which the unit will be stored on the class.	required
**pipe_config	PipeConfig	The pipe configuration for the new unit.	{}

Source code in pipeline/battery/cls.py

@classmethod
def add(cls, name: str, **pipe_config: Unpack[PipeConfig]) -> None:
    """
    Registers a new BatteryUnit as a class attribute.

    Creates a `BatteryUnit` from the provided pipe configuration and
    assigns it to the class under the given name. This allows dynamically
    extending the Battery registry at application startup.

    Args:
        name (str): The attribute name under which the unit will be stored on the class.
        **pipe_config (PipeConfig): The pipe configuration for the new unit.
    """
    setattr(cls, name, BatteryUnit(**pipe_config))

get(name) classmethod

Retrieves a registered BatteryUnit by name.

Parameters:

Name	Type	Description	Default
name	str	The attribute name of the unit to retrieve.	required

Returns:

Name	Type	Description
BatteryUnit	BatteryUnit	The BatteryUnit instance registered under the given name.

Raises:

Type	Description
AttributeError	If no unit with the given name has been registered on the class.

Source code in pipeline/battery/cls.py

@classmethod
def get(cls, name: str) -> BatteryUnit:
    """
    Retrieves a registered BatteryUnit by name.

    Args:
        name (str): The attribute name of the unit to retrieve.

    Returns:
        BatteryUnit: The BatteryUnit instance registered under the given name.

    Raises:
        AttributeError: If no unit with the given name has been registered on the class.
    """
    return getattr(cls, name)

use_email(min_length=6, max_length=64) staticmethod

Creates a BatteryUnit for email address validation.

Produces a unit that validates a string value as a well-formed email address within a configurable length range.

Parameters:

Name	Type	Description	Default
min_length	int	The minimum allowed length of the email address. Defaults to 6.	6
max_length	int	The maximum allowed length of the email address. Defaults to 64.	64

Returns:

Name	Type	Description
BatteryUnit	BatteryUnit	A unit for email validation.

Source code in pipeline/battery/cls.py

@staticmethod
def use_email(min_length: int = 6, max_length: int = 64) -> BatteryUnit:
    """
    Creates a BatteryUnit for email address validation.

    Produces a unit that validates a string value as a well-formed email
    address within a configurable length range.

    Args:
        min_length (int): The minimum allowed length of the email address.
            Defaults to 6.
        max_length (int): The maximum allowed length of the email address.
            Defaults to 64.

    Returns:
        BatteryUnit: A unit for email validation.
    """
    return BatteryUnit(
        type=str,
        conditions={
            Condition.MinLength: min_length,
            Condition.MaxLength: max_length
        },
        matches={Match.Format.Email: None},
        transform={Transform.Lowercase: None}
    )

use_limit(min_number=1, max_number=1000) staticmethod

Creates a BatteryUnit for pagination limit validation.

Produces a unit that validates an integer value representing the number of items to return per page, within a configurable range.

Parameters:

Name	Type	Description	Default
min_number	int	The minimum allowed limit value. Defaults to 1.	1
max_number	int	The maximum allowed limit value. Defaults to 1000.	1000

Returns:

Name	Type	Description
BatteryUnit	BatteryUnit	A unit for pagination limit validation.

Source code in pipeline/battery/cls.py

@staticmethod
def use_limit(min_number: int = 1, max_number: int = 1000) -> BatteryUnit:
    """
    Creates a BatteryUnit for pagination limit validation.

    Produces a unit that validates an integer value representing the
    number of items to return per page, within a configurable range.

    Args:
        min_number (int): The minimum allowed limit value. Defaults to 1.
        max_number (int): The maximum allowed limit value. Defaults to 1000.

    Returns:
        BatteryUnit: A unit for pagination limit validation.
    """
    return BatteryUnit(
        type=int,
        conditions={
            Condition.MinNumber: min_number,
            Condition.MaxNumber: max_number
        }
    )

use_offset(min_number=0, max_number=1000) staticmethod

Creates a BatteryUnit for pagination offset validation.

Produces a unit that validates an integer value representing the starting position of a result set, within a configurable range. Differs from use_limit only in its default min_number of 0, which allows an offset of zero as a valid starting point.

Parameters:

Name	Type	Description	Default
min_number	int	The minimum allowed offset value. Defaults to 0.	0
max_number	int	The maximum allowed offset value. Defaults to 1000.	1000

Returns:

Name	Type	Description
BatteryUnit	BatteryUnit	A unit for pagination offset validation.

Source code in pipeline/battery/cls.py

@staticmethod
def use_offset(min_number: int = 0, max_number: int = 1000) -> BatteryUnit:
    """
    Creates a BatteryUnit for pagination offset validation.

    Produces a unit that validates an integer value representing the
    starting position of a result set, within a configurable range.
    Differs from `use_limit` only in its default `min_number` of 0,
    which allows an offset of zero as a valid starting point.

    Args:
        min_number (int): The minimum allowed offset value. Defaults to 0.
        max_number (int): The maximum allowed offset value. Defaults to 1000.

    Returns:
        BatteryUnit: A unit for pagination offset validation.
    """
    return BatteryUnit(
        type=int,
        conditions={
            Condition.MinNumber: min_number,
            Condition.MaxNumber: max_number
        }
    )

use_password(policy) staticmethod

Creates a BatteryUnit for password validation.

Produces a unit that validates a string's length and, optionally, its complexity against predefined security patterns.

Parameters:

Name	Type	Description	Default
policy	Literal['relaxed', 'normal', 'strict']	The complexity level. If None, only length is validated (min: 6, max: 64).	required

Returns:

Name	Type	Description
BatteryUnit	BatteryUnit	A unit for password validation.

Source code in pipeline/battery/cls.py

@staticmethod
def use_password(
    policy: Literal["relaxed", "normal", "strict"] | None
) -> BatteryUnit:
    """
    Creates a BatteryUnit for password validation.

    Produces a unit that validates a string's length and, optionally, 
    its complexity against predefined security patterns.

    Args:
        policy (Literal["relaxed", "normal", "strict"]): The complexity level. If None, only
            length is validated (min: 6, max: 64).

    Returns:
        BatteryUnit: A unit for password validation.
    """
    config: PipeConfig = {
        'type': str,
        'conditions': {
            Condition.MinLength: 6,
            Condition.MaxLength: 64
        }
    }

    if policy is not None:
        config['matches'] = {Match.Format.Password: policy}

    return BatteryUnit(**config)

use_token(exact_length=32, digits_only=False) staticmethod

Creates a BatteryUnit for token validation.

Produces a unit that validates a string as a fixed-length token. Can be configured for alphanumeric characters or strictly digits.

Parameters:

Name	Type	Description	Default
exact_length	int	The required exact length of the token. Defaults to 32.	32
digits_only	bool	If True, validates using Match.Text.Digits. If False, validates using Match.Text.Alphanumeric. Defaults to False.	False

Returns:

Name	Type	Description
BatteryUnit	BatteryUnit	A unit for token validation.

Source code in pipeline/battery/cls.py

@staticmethod
def use_token(
    exact_length: int = 32, digits_only: bool = False
) -> BatteryUnit:
    """
    Creates a BatteryUnit for token validation.

    Produces a unit that validates a string as a fixed-length token.
    Can be configured for alphanumeric characters or strictly digits.

    Args:
        exact_length (int): The required exact length of the token. 
            Defaults to 32.
        digits_only (bool): If True, validates using Match.Text.Digits.
            If False, validates using Match.Text.Alphanumeric. Defaults to False.

    Returns:
        BatteryUnit: A unit for token validation.
    """
    pattern = Match.Text.Digits if digits_only else Match.Text.Alphanumeric

    return BatteryUnit(
        type=str,
        conditions={Condition.ExactLength: exact_length},
        matches={pattern: None}
    )

battery.unit

BatteryUnit

A reusable container for a single pipe configuration.

Stores a pre-defined set of pipe arguments that can be retrieved and optionally extended at the point of use via the sink() method.

Source code in pipeline/battery/unit.py

class BatteryUnit:
    """
    A reusable container for a single pipe configuration.

    Stores a pre-defined set of pipe arguments that can be retrieved and optionally extended at
    the point of use via the `sink()` method.
    """
    def __init__(self, **pipe_config: Unpack[PipeConfig]) -> None:
        """
        Initializes the BatteryUnit with a fixed pipe configuration.

        Args:
            **pipe_config (PipeConfig): The pipe configuration to store.
        """
        self.pipe_config: PipeConfig = pipe_config

    def sink(self, **update: Unpack[PipeUpdateConfig]) -> PipeConfig:
        """
        Returns the stored pipe configuration, optionally merged with overrides.

        This method is the primary way to consume a BatteryUnit. It returns a
        `PipeConfig` dictionary ready to be passed to a `Pipe` or `Pipeline`.   
        Any keyword arguments provided in `update` will overwrite the
        corresponding keys in the stored configuration, allowing for
        field-specific customization without modifying the original unit.

        Args:
            **update (PipeUpdateConfig): Optional overrides to merge into the stored configuration.

        Returns:
            PipeConfig: A merged dictionary of the stored configuration and any provided overrides.
        """
        if update is None:
            return self.pipe_config

        return {**self.pipe_config, **update}

__init__(**pipe_config)

Initializes the BatteryUnit with a fixed pipe configuration.

Parameters:

Name	Type	Description	Default
**pipe_config	PipeConfig	The pipe configuration to store.	{}

Source code in pipeline/battery/unit.py

def __init__(self, **pipe_config: Unpack[PipeConfig]) -> None:
    """
    Initializes the BatteryUnit with a fixed pipe configuration.

    Args:
        **pipe_config (PipeConfig): The pipe configuration to store.
    """
    self.pipe_config: PipeConfig = pipe_config

sink(**update)

Returns the stored pipe configuration, optionally merged with overrides.

This method is the primary way to consume a BatteryUnit. It returns a PipeConfig dictionary ready to be passed to a Pipe or Pipeline.
Any keyword arguments provided in update will overwrite the corresponding keys in the stored configuration, allowing for field-specific customization without modifying the original unit.

Parameters:

Name	Type	Description	Default
**update	PipeUpdateConfig	Optional overrides to merge into the stored configuration.	{}

Returns:

Name	Type	Description
PipeConfig	PipeConfig	A merged dictionary of the stored configuration and any provided overrides.

Source code in pipeline/battery/unit.py

def sink(self, **update: Unpack[PipeUpdateConfig]) -> PipeConfig:
    """
    Returns the stored pipe configuration, optionally merged with overrides.

    This method is the primary way to consume a BatteryUnit. It returns a
    `PipeConfig` dictionary ready to be passed to a `Pipe` or `Pipeline`.   
    Any keyword arguments provided in `update` will overwrite the
    corresponding keys in the stored configuration, allowing for
    field-specific customization without modifying the original unit.

    Args:
        **update (PipeUpdateConfig): Optional overrides to merge into the stored configuration.

    Returns:
        PipeConfig: A merged dictionary of the stored configuration and any provided overrides.
    """
    if update is None:
        return self.pipe_config

    return {**self.pipe_config, **update}