Skip to content

Condition Stage

The Condition stage is the first step in the Pipe lifecycle. It is responsible for validating the fundamental integrity of the data, such as its type, presence, size, and basic constraints.

Execution Rule

If a condition fails, the pipe execution will not stop immediately. It will evaluate every condition to gather all errors. However, certain conditions have a special flag ConditionFlag.BREAK_PIPE_LOOP_ON_ERROR (such as Condition.ValueType) that breaks the loop upon an error.

Technical Reference

The following section is automatically generated from the source code, detailing the available condition handlers and their configurations.

Condition

Registry for all condition handlers.

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

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

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

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

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Invalid format. Please provide a {self.argument.__name__}."
        }

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

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

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Too short. This must be at least {self.argument} characters."
                if isinstance(self.value, str) else
                f"Please select at least {self.argument} items."
        }

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

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

        SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"Too long. Maximum length is {self.argument} characters."
                if isinstance(self.value, str) else
                f"You can select a maximum of {self.argument} items."
        }

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

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

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self: f"Must be {self.argument} or greater."
        }

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

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

        ERROR_TEMPLATES = {
            HandlerMode.ROOT: lambda self: f"Must be {self.argument} or less."
        }

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

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

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

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

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

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self:
                f"This value is not allowed. Please use a different value that is not in {list(self.argument)}."
        }

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

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

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self: f"The value must match {self.argument}."
        }

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

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

        ERROR_TEMPLATES = {
            HandlerMode.ROOT:
                lambda self: f"Value cannot be equal to {self.argument}."
        }

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

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

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

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

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

        ERROR_TEMPLATES = {
            HandlerMode.CONTEXT:
                lambda self:
                f"This cannot be the same as the {self.input_argument} field."
        }

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

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

        SUPPORT = (HandlerMode.ROOT, )

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

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

            return self.metadata['errors'] is None

DoesNotMatchField

Bases: ConditionHandler[Any, Any]

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

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

    ERROR_TEMPLATES = {
        HandlerMode.CONTEXT:
            lambda self:
            f"This cannot be the same as the {self.input_argument} field."
    }

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

Equal

Bases: ConditionHandler[Any, Any]

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

Source code in pipeline/handlers/condition_handler/condition.py 
121
122
123
124
125
126
127
128
129
130
131
class Equal(ConditionHandler[Any, Any]):
    """Ensures the value is strictly equal to the argument or a specific context field"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self: f"The value must match {self.argument}."
    }

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

IncludedIn

Bases: ConditionHandler[Any, Iterable]

Ensures the value exists within the provided Iterable

Source code in pipeline/handlers/condition_handler/condition.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class IncludedIn(ConditionHandler[Any, Iterable]):
    """Ensures the value exists within the provided Iterable"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

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

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

MatchesField

Bases: ConditionHandler[Any, Any]

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

Source code in pipeline/handlers/condition_handler/condition.py 
145
146
147
148
149
150
151
152
153
154
155
class MatchesField(ConditionHandler[Any, Any]):
    """Validates that the current value matches the value of another field in the context (e.g., password confirmation)"""
    SUPPORT = (HandlerMode.CONTEXT, )

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

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

MaxLength

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

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

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

    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Too long. Maximum length is {self.argument} characters."
            if isinstance(self.value, str) else
            f"You can select a maximum of {self.argument} items."
    }

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

MaxNumber

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

Ensures the numeric value is less than or equal to N

Source code in pipeline/handlers/condition_handler/condition.py 
84
85
86
87
88
89
90
91
92
93
class MaxNumber(ConditionHandler[int | float, int | float]):
    """Ensures the numeric value is less than or equal to N"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT: lambda self: f"Must be {self.argument} or less."
    }

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

MinLength

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

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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Too short. This must be at least {self.argument} characters."
            if isinstance(self.value, str) else
            f"Please select at least {self.argument} items."
    }

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

MinNumber

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

Ensures the numeric value is greater than or equal to N

Source code in pipeline/handlers/condition_handler/condition.py 
72
73
74
75
76
77
78
79
80
81
82
class MinNumber(ConditionHandler[int | float, int | float]):
    """Ensures the numeric value is greater than or equal to N"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM, HandlerMode.CONTEXT)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self: f"Must be {self.argument} or greater."
    }

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

NotEqual

Bases: ConditionHandler[Any, Any]

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

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

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self: f"Value cannot be equal to {self.argument}."
    }

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

NotIncludedIn

Bases: ConditionHandler[Any, Iterable]

Ensures the value does not exist within the provided blacklist

Source code in pipeline/handlers/condition_handler/condition.py 
108
109
110
111
112
113
114
115
116
117
118
119
class NotIncludedIn(ConditionHandler[Any, Iterable]):
    """Ensures the value does not exist within the provided blacklist"""
    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"This value is not allowed. Please use a different value that is not in {list(self.argument)}."
    }

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

Pipeline

Bases: ConditionHandler[dict, Pipeline]

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

Source code in pipeline/handlers/condition_handler/condition.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class Pipeline(ConditionHandler[dict, Pipeline]):
    """Validates a dictionary using the same rules as the normal pipeline, but for nested data."""
    FLAGS = (ConditionFlag.RETURN_ONLY_ERROR_MSG, )

    SUPPORT = (HandlerMode.ROOT, )

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

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

        return self.metadata['errors'] is None

ValueType

Bases: ConditionHandler[Any, type]

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

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

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

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

    SUPPORT = (HandlerMode.ROOT, HandlerMode.ITEM)

    ERROR_TEMPLATES = {
        HandlerMode.ROOT:
            lambda self:
            f"Invalid format. Please provide a {self.argument.__name__}."
    }

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