Skip to content

validator

The validation base class.

Validator

Bases: Serializable

Class that represents a validation, including condition, and results for success or failure.

Source code in mlte/validation/validator.py
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
class Validator(Serializable):
    """
    Class that represents a validation, including condition, and results for success or failure.
    """

    def __init__(
        self,
        *,
        bool_exp: Optional[Callable[[Any], bool]] = None,
        thresholds: list[str] = [],
        success: Optional[str] = None,
        failure: Optional[str] = None,
        default_success: str = "",
        default_failure: str = "",
        info: Optional[str] = None,
        input_types: list[str] = [],
        creator: Optional[FunctionInfo] = None,
    ):
        """
        Constructor.

        :param bool_exp: A boolean expression that can be used to test the actual condition we want to validate.
        :param thresholds: A list of serialized thesholds used in the bool exp, for auditing purposes.
        :param success: A string indicating the message to record in case of success (bool_exp evaluating to True).
        :param failure: A string indicating the message to record in case of failure (bool_exp evaluating to False).
        :param default_success: A more generic value to use if success is not set.
        :param default_failure: A more generic value to use if failure is not set.
        :param info: A string indicating the message to record in case no bool expression is passed (no condition, just recording information).
        :param input_types: A list of strings indicating the type of input the validator will expect.
        :param creator: Information about the class and method that created this validator, if any.
        """
        # Use default if needed.
        if not success:
            success = default_success if default_success else None
        if not failure:
            failure = default_failure if default_failure else None

        if success is not None and failure is None:
            raise ValueError(
                "If success message is defined, failure message has to be defined as well"
            )
        if success is None and failure is not None:
            raise ValueError(
                "If failure message is defined, success message has to be defined as well"
            )
        if success is None and failure is None and info is None:
            raise ValueError(
                "All messages can't be empty, either info or success and failure have to be defined"
            )

        self.bool_exp = bool_exp
        self.thresholds = thresholds.copy()
        self.success = success
        self.failure = failure
        self.default_success = default_success
        self.default_failure = default_failure
        self.info = info
        self.input_types = input_types
        self.creator = creator

        self.bool_exp_str = (
            reflection.get_lambda_code(bool_exp)
            if bool_exp is not None
            else None
        )
        """We also store the bool expression as a string from its code, for tracking purposes."""

    @staticmethod
    def build_validator(
        bool_exp: Optional[Callable[[Any], bool]] = None,
        thresholds: list[Any] = [],
        success: Optional[str] = None,
        failure: Optional[str] = None,
        default_success: str = "",
        default_failure: str = "",
        info: Optional[str] = None,
        input_types: list[type] = [Evidence],
    ) -> Validator:
        """
        Creates a Validator using the provided test, extracting context info from the function that called us.

        :param bool_exp: A boolean expression that can be used to test the actual condition we want to validate.
        :param thresholds: A list of thesholds used in the bool exp, potentially unit-aware, for auditing purposes.
        :param success: A string indicating the message to record in case of success (bool_exp evaluating to True).
        :param failure: A string indicating the message to record in case of failure (bool_exp evaluating to False).
        :param default_success: A more generic value to use if success is not set.
        :param default_failure: A more generic value to use if failure is not set.
        :param info: A string indicating the message to record in case no bool expression is passed (no condition, just recording information).
        :param input_types: A list of types indicating the type of input the validator will expect.
        :returns: A Validator, potentially with caller creator information.
        """
        # Get function info, passing our caller as argument.
        curr_frame = inspect.currentframe()
        caller_function = curr_frame.f_back if curr_frame is not None else None
        function_info = FunctionInfo.get_function_info(caller_function)

        # Build the validator. We can't really check at this point if the bool_exp actually returns a bool.
        validator = Validator(
            bool_exp=bool_exp,
            thresholds=[str(threshold) for threshold in thresholds],
            success=success,
            failure=failure,
            default_success=default_success,
            default_failure=default_failure,
            info=info,
            input_types=[
                meta.get_qualified_name(input_type)
                for input_type in input_types
            ],
            creator=function_info,
        )
        return validator

    @staticmethod
    def build_info_validator(info: str) -> Validator:
        """
        Creates a genering Validator with no bool exp and with the given info.
        Useful for validations that can't be automated but need to be recorded.
        :param info: The information to record.
        :return: The Validator that can be used.
        """
        validator: Validator = Validator.build_validator(
            info=info, input_types=[]
        )
        return validator

    def validate(self, *args, **kwargs) -> Result:
        """
        Generates a result based on the arguments received, and the configured attributes in the Validator.

        :param args, kwargs: Arguments to pass to the boolean expression to be evaluated in this specific case.
        :return: A Result, including a message with details about the validation result.
        """
        if self.bool_exp is None and self.info is None:
            raise RuntimeError(
                "Can't validate, Validator has no bool expression and is also missing informational message that is used in those cases."
            )

        # Check we got proper arguments.
        self._check_arguments(*args, **kwargs)

        # First execute bool expression (if any), and get its boolean result.
        executed_bool_exp_value: Optional[bool] = None
        if self.bool_exp is not None:
            executed_bool_exp_value = self.bool_exp(*args, **kwargs)
            if not isinstance(executed_bool_exp_value, bool):
                raise ValueError(
                    "Configured bool expression does not return a bool."
                )

        # Create the result to be returned.
        values_str = self._args_to_string(*args, **kwargs)
        result = (
            Info(self.info)
            if self.bool_exp is None and self.info is not None
            else (
                Success(
                    f"{self.success}",
                    additional_data=f"{self.default_success}{values_str}",
                )
                if executed_bool_exp_value
                else Failure(
                    f"{self.failure}",
                    additional_data=f"{self.default_failure}{values_str}",
                )
            )
        )
        return result

    def _check_arguments(self, *args, **kwargs):
        """Checks that the received arguments are of the expected types."""
        # Since input types is an ordered list, we assume we get them ordered as well, or we have no way to check.
        all_arguments = [arg for arg in args]
        all_arguments.extend([value for _, value in kwargs.items()])

        if len(self.input_types) == 0:
            # Input types are optional, ignore check if they were not configured or there aren't any.
            return

        if len(all_arguments) != len(self.input_types):
            raise RuntimeError(
                f"Invalid amoung of arguments: validator expected {len(self.input_types)}, but got {len(all_arguments)}"
            )

        for input_type in self.input_types:
            for arg in all_arguments:
                if input_type != meta.get_qualified_name(type(arg)):
                    raise RuntimeError(
                        f"Invalid argument type received: expected {input_type}, received {meta.get_qualified_name(type(arg))}"
                    )

    def _args_to_string(self, *args, **kwargs) -> str:
        """
        Stringify arguments so that result's message can include generic information about arguments used when validating.
        """
        # First ensure args are turned to string separately, to allow them to use their own str()
        MAX_STRING_LENGTH = 300
        str_args = [
            (
                str(arg)
                if len(str(arg)) < MAX_STRING_LENGTH
                else str(arg)[:MAX_STRING_LENGTH] + "..."
            )
            for arg in args
        ]
        str_kwargs = {key: str(arg) for key, arg in kwargs.items()}

        # Now string them together, depending on whether we got args of each type.
        values = " - values: "
        if len(args) > 0:
            values = values + f"{json.dumps(str_args)}"
        if len(args) > 0 and len(kwargs) > 0:
            values = values + f"{', '}"
        if len(kwargs) > 0:
            values = values + f"{json.dumps(str_kwargs)}"

        return values

    # -------------------------------------------------------------------------
    # Model handling.
    # -------------------------------------------------------------------------

    def to_model(self) -> ValidatorModel:
        """
        Returns this validator as a model.

        :return: The serialized model object.
        """
        return ValidatorModel(
            bool_exp=(
                serializing.encode_callable(self.bool_exp)
                if self.bool_exp
                else None
            ),
            thresholds=self.thresholds,
            success=self.success,
            failure=self.failure,
            default_success=self.default_success,
            default_failure=self.default_failure,
            info=self.info,
            bool_exp_str=self.bool_exp_str,
            input_types=self.input_types,
            creator_entity=(
                self.creator.function_parent
                if self.creator is not None
                else None
            ),
            creator_function=(
                self.creator.function_name if self.creator is not None else None
            ),
            creator_args=(
                [str(arg) for arg in self.creator.arguments]
                if self.creator is not None
                else []
            ),
        )

    @classmethod
    def from_model(cls, model: BaseModel) -> Validator:
        """
        Deserialize a Validator from a model.

        :param model: The model.

        :return: The deserialized Validator
        """
        model = typing.cast(ValidatorModel, model)
        validator: Validator = Validator(
            bool_exp=(
                typing.cast(
                    Callable[[Any], bool],
                    serializing.decode_callable(model.bool_exp),
                )
                if model.bool_exp
                else None
            ),
            thresholds=model.thresholds,
            success=model.success,
            failure=model.failure,
            default_success=model.default_success,
            default_failure=model.default_failure,
            info=model.info,
            input_types=model.input_types,
            creator=(
                FunctionInfo(
                    model.creator_function,
                    model.creator_args,
                    model.creator_entity,
                )
                if model.creator_function is not None
                and model.creator_entity is not None
                else None
            ),
        )
        return validator

    # -------------------------------------------------------------------------
    # Equality Testing
    # -------------------------------------------------------------------------

    def __eq__(self, other: object) -> bool:
        """Test instance for equality."""
        if not isinstance(other, Validator):
            return False

        # Compare bool exp functions separately, then remove from model as encoding can be different due to creation context.
        bool_exps_same = (
            serializing.compare_callable(self.bool_exp, other.bool_exp)
            if self.bool_exp and other.bool_exp
            else self.bool_exp == other.bool_exp
        )

        a_model = self.to_model()
        b_model = other.to_model()

        a_model.bool_exp = None
        b_model.bool_exp = None

        return a_model == b_model and bool_exps_same

bool_exp_str = reflection.get_lambda_code(bool_exp) if bool_exp is not None else None instance-attribute

We also store the bool expression as a string from its code, for tracking purposes.

__eq__(other)

Test instance for equality.

Source code in mlte/validation/validator.py
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def __eq__(self, other: object) -> bool:
    """Test instance for equality."""
    if not isinstance(other, Validator):
        return False

    # Compare bool exp functions separately, then remove from model as encoding can be different due to creation context.
    bool_exps_same = (
        serializing.compare_callable(self.bool_exp, other.bool_exp)
        if self.bool_exp and other.bool_exp
        else self.bool_exp == other.bool_exp
    )

    a_model = self.to_model()
    b_model = other.to_model()

    a_model.bool_exp = None
    b_model.bool_exp = None

    return a_model == b_model and bool_exps_same

__init__(*, bool_exp=None, thresholds=[], success=None, failure=None, default_success='', default_failure='', info=None, input_types=[], creator=None)

Constructor.

Parameters:

Name Type Description Default
bool_exp Optional[Callable[[Any], bool]]

A boolean expression that can be used to test the actual condition we want to validate.

None
thresholds list[str]

A list of serialized thesholds used in the bool exp, for auditing purposes.

[]
success Optional[str]

A string indicating the message to record in case of success (bool_exp evaluating to True).

None
failure Optional[str]

A string indicating the message to record in case of failure (bool_exp evaluating to False).

None
default_success str

A more generic value to use if success is not set.

''
default_failure str

A more generic value to use if failure is not set.

''
info Optional[str]

A string indicating the message to record in case no bool expression is passed (no condition, just recording information).

None
input_types list[str]

A list of strings indicating the type of input the validator will expect.

[]
creator Optional[FunctionInfo]

Information about the class and method that created this validator, if any.

None
Source code in mlte/validation/validator.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    *,
    bool_exp: Optional[Callable[[Any], bool]] = None,
    thresholds: list[str] = [],
    success: Optional[str] = None,
    failure: Optional[str] = None,
    default_success: str = "",
    default_failure: str = "",
    info: Optional[str] = None,
    input_types: list[str] = [],
    creator: Optional[FunctionInfo] = None,
):
    """
    Constructor.

    :param bool_exp: A boolean expression that can be used to test the actual condition we want to validate.
    :param thresholds: A list of serialized thesholds used in the bool exp, for auditing purposes.
    :param success: A string indicating the message to record in case of success (bool_exp evaluating to True).
    :param failure: A string indicating the message to record in case of failure (bool_exp evaluating to False).
    :param default_success: A more generic value to use if success is not set.
    :param default_failure: A more generic value to use if failure is not set.
    :param info: A string indicating the message to record in case no bool expression is passed (no condition, just recording information).
    :param input_types: A list of strings indicating the type of input the validator will expect.
    :param creator: Information about the class and method that created this validator, if any.
    """
    # Use default if needed.
    if not success:
        success = default_success if default_success else None
    if not failure:
        failure = default_failure if default_failure else None

    if success is not None and failure is None:
        raise ValueError(
            "If success message is defined, failure message has to be defined as well"
        )
    if success is None and failure is not None:
        raise ValueError(
            "If failure message is defined, success message has to be defined as well"
        )
    if success is None and failure is None and info is None:
        raise ValueError(
            "All messages can't be empty, either info or success and failure have to be defined"
        )

    self.bool_exp = bool_exp
    self.thresholds = thresholds.copy()
    self.success = success
    self.failure = failure
    self.default_success = default_success
    self.default_failure = default_failure
    self.info = info
    self.input_types = input_types
    self.creator = creator

    self.bool_exp_str = (
        reflection.get_lambda_code(bool_exp)
        if bool_exp is not None
        else None
    )
    """We also store the bool expression as a string from its code, for tracking purposes."""

build_info_validator(info) staticmethod

Creates a genering Validator with no bool exp and with the given info. Useful for validations that can't be automated but need to be recorded.

Parameters:

Name Type Description Default
info str

The information to record.

required

Returns:

Type Description
Validator

The Validator that can be used.

Source code in mlte/validation/validator.py
132
133
134
135
136
137
138
139
140
141
142
143
@staticmethod
def build_info_validator(info: str) -> Validator:
    """
    Creates a genering Validator with no bool exp and with the given info.
    Useful for validations that can't be automated but need to be recorded.
    :param info: The information to record.
    :return: The Validator that can be used.
    """
    validator: Validator = Validator.build_validator(
        info=info, input_types=[]
    )
    return validator

build_validator(bool_exp=None, thresholds=[], success=None, failure=None, default_success='', default_failure='', info=None, input_types=[Evidence]) staticmethod

Creates a Validator using the provided test, extracting context info from the function that called us.

Parameters:

Name Type Description Default
bool_exp Optional[Callable[[Any], bool]]

A boolean expression that can be used to test the actual condition we want to validate.

None
thresholds list[Any]

A list of thesholds used in the bool exp, potentially unit-aware, for auditing purposes.

[]
success Optional[str]

A string indicating the message to record in case of success (bool_exp evaluating to True).

None
failure Optional[str]

A string indicating the message to record in case of failure (bool_exp evaluating to False).

None
default_success str

A more generic value to use if success is not set.

''
default_failure str

A more generic value to use if failure is not set.

''
info Optional[str]

A string indicating the message to record in case no bool expression is passed (no condition, just recording information).

None
input_types list[type]

A list of types indicating the type of input the validator will expect.

[Evidence]

Returns:

Type Description
Validator

A Validator, potentially with caller creator information.

Source code in mlte/validation/validator.py
 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
@staticmethod
def build_validator(
    bool_exp: Optional[Callable[[Any], bool]] = None,
    thresholds: list[Any] = [],
    success: Optional[str] = None,
    failure: Optional[str] = None,
    default_success: str = "",
    default_failure: str = "",
    info: Optional[str] = None,
    input_types: list[type] = [Evidence],
) -> Validator:
    """
    Creates a Validator using the provided test, extracting context info from the function that called us.

    :param bool_exp: A boolean expression that can be used to test the actual condition we want to validate.
    :param thresholds: A list of thesholds used in the bool exp, potentially unit-aware, for auditing purposes.
    :param success: A string indicating the message to record in case of success (bool_exp evaluating to True).
    :param failure: A string indicating the message to record in case of failure (bool_exp evaluating to False).
    :param default_success: A more generic value to use if success is not set.
    :param default_failure: A more generic value to use if failure is not set.
    :param info: A string indicating the message to record in case no bool expression is passed (no condition, just recording information).
    :param input_types: A list of types indicating the type of input the validator will expect.
    :returns: A Validator, potentially with caller creator information.
    """
    # Get function info, passing our caller as argument.
    curr_frame = inspect.currentframe()
    caller_function = curr_frame.f_back if curr_frame is not None else None
    function_info = FunctionInfo.get_function_info(caller_function)

    # Build the validator. We can't really check at this point if the bool_exp actually returns a bool.
    validator = Validator(
        bool_exp=bool_exp,
        thresholds=[str(threshold) for threshold in thresholds],
        success=success,
        failure=failure,
        default_success=default_success,
        default_failure=default_failure,
        info=info,
        input_types=[
            meta.get_qualified_name(input_type)
            for input_type in input_types
        ],
        creator=function_info,
    )
    return validator

from_model(model) classmethod

Deserialize a Validator from a model.

Parameters:

Name Type Description Default
model BaseModel

The model.

required

Returns:

Type Description
Validator

The deserialized Validator

Source code in mlte/validation/validator.py
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
@classmethod
def from_model(cls, model: BaseModel) -> Validator:
    """
    Deserialize a Validator from a model.

    :param model: The model.

    :return: The deserialized Validator
    """
    model = typing.cast(ValidatorModel, model)
    validator: Validator = Validator(
        bool_exp=(
            typing.cast(
                Callable[[Any], bool],
                serializing.decode_callable(model.bool_exp),
            )
            if model.bool_exp
            else None
        ),
        thresholds=model.thresholds,
        success=model.success,
        failure=model.failure,
        default_success=model.default_success,
        default_failure=model.default_failure,
        info=model.info,
        input_types=model.input_types,
        creator=(
            FunctionInfo(
                model.creator_function,
                model.creator_args,
                model.creator_entity,
            )
            if model.creator_function is not None
            and model.creator_entity is not None
            else None
        ),
    )
    return validator

to_model()

Returns this validator as a model.

Returns:

Type Description
ValidatorModel

The serialized model object.

Source code in mlte/validation/validator.py
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def to_model(self) -> ValidatorModel:
    """
    Returns this validator as a model.

    :return: The serialized model object.
    """
    return ValidatorModel(
        bool_exp=(
            serializing.encode_callable(self.bool_exp)
            if self.bool_exp
            else None
        ),
        thresholds=self.thresholds,
        success=self.success,
        failure=self.failure,
        default_success=self.default_success,
        default_failure=self.default_failure,
        info=self.info,
        bool_exp_str=self.bool_exp_str,
        input_types=self.input_types,
        creator_entity=(
            self.creator.function_parent
            if self.creator is not None
            else None
        ),
        creator_function=(
            self.creator.function_name if self.creator is not None else None
        ),
        creator_args=(
            [str(arg) for arg in self.creator.arguments]
            if self.creator is not None
            else []
        ),
    )

validate(*args, **kwargs)

Generates a result based on the arguments received, and the configured attributes in the Validator.

Parameters:

Name Type Description Default
kwargs (args,)

Arguments to pass to the boolean expression to be evaluated in this specific case.

{}

Returns:

Type Description
Result

A Result, including a message with details about the validation result.

Source code in mlte/validation/validator.py
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def validate(self, *args, **kwargs) -> Result:
    """
    Generates a result based on the arguments received, and the configured attributes in the Validator.

    :param args, kwargs: Arguments to pass to the boolean expression to be evaluated in this specific case.
    :return: A Result, including a message with details about the validation result.
    """
    if self.bool_exp is None and self.info is None:
        raise RuntimeError(
            "Can't validate, Validator has no bool expression and is also missing informational message that is used in those cases."
        )

    # Check we got proper arguments.
    self._check_arguments(*args, **kwargs)

    # First execute bool expression (if any), and get its boolean result.
    executed_bool_exp_value: Optional[bool] = None
    if self.bool_exp is not None:
        executed_bool_exp_value = self.bool_exp(*args, **kwargs)
        if not isinstance(executed_bool_exp_value, bool):
            raise ValueError(
                "Configured bool expression does not return a bool."
            )

    # Create the result to be returned.
    values_str = self._args_to_string(*args, **kwargs)
    result = (
        Info(self.info)
        if self.bool_exp is None and self.info is not None
        else (
            Success(
                f"{self.success}",
                additional_data=f"{self.default_success}{values_str}",
            )
            if executed_bool_exp_value
            else Failure(
                f"{self.failure}",
                additional_data=f"{self.default_failure}{values_str}",
            )
        )
    )
    return result