#253

Committed 14 Apr 2025 12:28PM UTC coverage: 98.409% (-0.4%) from 98.773%

Build # #253

Build Type

push

coveralls-python

Committed by

LeanderCS

Commit Message

Updated error handling and changed broad Exception to specific errors

Coverage Stats

42 of 50 new or added lines in 10 files covered. (84.0%)

1856 of 1886 relevant lines covered (98.41%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.45

/flask_inputfilter/InputFilter.py

from __future__ import annotations

import json
import logging
from typing import (
    Any,
    Callable,
    Dict,
    List,
    Optional,
    Tuple,
    Type,
    TypeVar,
    Union,
)

from flask import Response, g, request
from typing_extensions import final

from flask_inputfilter.Condition import BaseCondition
from flask_inputfilter.Exception import ValidationError
from flask_inputfilter.Filter import BaseFilter
from flask_inputfilter.Mixin import (
    ConditionMixin,
    DataMixin,
    ErrorHandlingMixin,
    ExternalApiMixin,
    FieldMixin,
    FilterMixin,
    ModelMixin,
    ValidationMixin,
)
from flask_inputfilter.Model import FieldModel
from flask_inputfilter.Validator import BaseValidator

T = TypeVar("T")


class InputFilter(
    ConditionMixin,
    DataMixin,
    ErrorHandlingMixin,
    ExternalApiMixin,
    FieldMixin,
    FilterMixin,
    ModelMixin,
    ValidationMixin,
):
    """
    Base class for input filters.
    """

    __slots__ = (
        "__methods",
        "_fields",
        "_conditions",
        "_global_filters",
        "_global_validators",
        "_data",
        "_validated_data",
        "_errors",
        "_model_class",
    )

    def __init__(self, methods: Optional[List[str]] = None) -> None:
        self.__methods: frozenset = frozenset(
            methods or ["GET", "POST", "PATCH", "PUT", "DELETE"]
        )
        self._fields: Dict[str, FieldModel] = {}
        self._conditions: List[BaseCondition] = []
        self._global_filters: List[BaseFilter] = []
        self._global_validators: List[BaseValidator] = []
        self._data: Dict[str, Any] = {}
        self._validated_data: Dict[str, Any] = {}
        self._errors: Dict[str, str] = {}
        self._model_class: Optional[Type[T]] = None

    @final
    def isValid(self) -> bool:
        """
        Checks if the object's state or its attributes meet certain
        conditions to be considered valid. This function is typically used to
        ensure that the current state complies with specific requirements or
        rules.

        Returns:
            bool: Returns True if the state or attributes of the object fulfill
                all required conditions; otherwise, returns False.
        """
        try:
            self.validateData()

        except ValidationError as e:
            self._errors = e.args[0]
            return False

        return True

    @classmethod
    @final
    def validate(
        cls,
    ) -> Callable[
        [Any],
        Callable[
            [Tuple[Any, ...], Dict[str, Any]],
            Union[Response, Tuple[Any, Dict[str, Any]]],
        ],
    ]:
        """
        Decorator for validating input data in routes.
        """

        def decorator(
            f,
        ) -> Callable[
            [Tuple[Any, ...], Dict[str, Any]],
            Union[Response, Tuple[Any, Dict[str, Any]]],
        ]:
            def wrapper(
                *args, **kwargs
            ) -> Union[Response, Tuple[Any, Dict[str, Any]]]:
                input_filter = cls()
                if request.method not in input_filter.__methods:
                    return Response(status=405, response="Method Not Allowed")

                data = request.json if request.is_json else request.args

                try:
                    kwargs = kwargs or {}

                    input_filter._data = {**data, **kwargs}

                    validated_data = input_filter.validateData()

                    if input_filter._model_class is not None:
                        g.validated_data = input_filter.serialize()
                    else:
                        g.validated_data = validated_data

                except ValidationError as e:
                    return Response(
                        status=400,
                        response=json.dumps(e.args[0]),
                        mimetype="application/json",
                    )

                except Exception:
                    logging.getLogger(__name__).exception(
                        "An unexpected exception occurred while "
                        "validating input data.",
                    )
                    return Response(status=500)

                return f(*args, **kwargs)

            return wrapper

        return decorator

    @final
    def validateData(
        self, data: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        Validates input data against defined field rules, including applying
        filters, validators, custom logic steps, and fallback mechanisms. The
        validation process also ensures the required fields are handled
        appropriately and conditions are checked after processing.

        Args:
            data (Dict[str, Any]): A dictionary containing the input data to
                be validated where keys represent field names and values
                represent the corresponding data.

        Returns:
            Dict[str, Any]: A dictionary containing the validated data with
                any modifications, default values, or processed values as
                per the defined validation rules.

        Raises:
            Any errors raised during external API calls, validation, or
                logical steps execution of the respective fields or conditions
                will propagate without explicit handling here.
        """
        data = data or self._data
        errors = {}

        for field_name, field_info in self._fields.items():
            value = data.get(field_name)

            required = field_info.required
            default = field_info.default
            fallback = field_info.fallback
            filters = field_info.filters
            validators = field_info.validators
            steps = field_info.steps
            external_api = field_info.external_api
            copy = field_info.copy

            try:
                if copy:
                    value = self._validated_data.get(copy)

                if external_api:
                    value = self._ExternalApiMixin__callExternalApi(
                        external_api, fallback, self._validated_data
                    )

                value = self._FilterMixin__applyFilters(filters, value)
                value = (
                    self._ValidationMixin__validateField(
                        validators, fallback, value
                    )
                    or value
                )
                value = (
                    self._FieldMixin__applySteps(steps, fallback, value)
                    or value
                )
                value = self._FieldMixin__checkForRequired(
                    field_name, required, default, fallback, value
                )

                self._validated_data[field_name] = value

            except ValidationError as e:
                errors[field_name] = str(e)

        try:
            self._ConditionMixin__checkConditions(self._validated_data)
        except ValidationError as e:
            errors["_condition"] = str(e)

        if errors:
            raise ValidationError(errors)

        return self._validated_data

1	from __future__ import annotations	1✔
2
3	import json	1✔
4	import logging	1✔
5	from typing import (	1✔
6	Any,
7	Callable,
8	Dict,
9	List,
10	Optional,
11	Tuple,
12	Type,
13	TypeVar,
14	Union,
15	)
16
17	from flask import Response, g, request	1✔
18	from typing_extensions import final	1✔
19
20	from flask_inputfilter.Condition import BaseCondition	1✔
21	from flask_inputfilter.Exception import ValidationError	1✔
22	from flask_inputfilter.Filter import BaseFilter	1✔
23	from flask_inputfilter.Mixin import (	1✔
24	ConditionMixin,
25	DataMixin,
26	ErrorHandlingMixin,
27	ExternalApiMixin,
28	FieldMixin,
29	FilterMixin,
30	ModelMixin,
31	ValidationMixin,
32	)
33	from flask_inputfilter.Model import FieldModel	1✔
34	from flask_inputfilter.Validator import BaseValidator	1✔
35
36	T = TypeVar("T")	1✔
37
38
39	class InputFilter(	1✔
40	ConditionMixin,
41	DataMixin,
42	ErrorHandlingMixin,
43	ExternalApiMixin,
44	FieldMixin,
45	FilterMixin,
46	ModelMixin,
47	ValidationMixin,
48	):
49	"""
50	Base class for input filters.
51	"""
52
53	__slots__ = (	1✔
54	"__methods",
55	"_fields",
56	"_conditions",
57	"_global_filters",
58	"_global_validators",
59	"_data",
60	"_validated_data",
61	"_errors",
62	"_model_class",
63	)
64
65	def __init__(self, methods: Optional[List[str]] = None) -> None:	1✔
66	self.__methods: frozenset = frozenset(	1✔
67	methods or ["GET", "POST", "PATCH", "PUT", "DELETE"]
68	)
69	self._fields: Dict[str, FieldModel] = {}	1✔
70	self._conditions: List[BaseCondition] = []	1✔
71	self._global_filters: List[BaseFilter] = []	1✔
72	self._global_validators: List[BaseValidator] = []	1✔
73	self._data: Dict[str, Any] = {}	1✔
74	self._validated_data: Dict[str, Any] = {}	1✔
75	self._errors: Dict[str, str] = {}	1✔
76	self._model_class: Optional[Type[T]] = None	1✔
77
78	@final	1✔
79	def isValid(self) -> bool:
80	"""
81	Checks if the object's state or its attributes meet certain
82	conditions to be considered valid. This function is typically used to
83	ensure that the current state complies with specific requirements or
84	rules.
85
86	Returns:
87	bool: Returns True if the state or attributes of the object fulfill
88	all required conditions; otherwise, returns False.
89	"""
90	try:	1✔
91	self.validateData()	1✔
92
93	except ValidationError as e:	1✔
94	self._errors = e.args[0]	1✔
95	return False	1✔
96
97	return True	1✔
98
99	@classmethod	1✔
100	@final	1✔
101	def validate(
102	cls,
103	) -> Callable[
104	[Any],
105	Callable[
106	[Tuple[Any, ...], Dict[str, Any]],
107	Union[Response, Tuple[Any, Dict[str, Any]]],
108	],
109	]:
110	"""
111	Decorator for validating input data in routes.
112	"""
113
114	def decorator(	1✔
115	f,
116	) -> Callable[
117	[Tuple[Any, ...], Dict[str, Any]],
118	Union[Response, Tuple[Any, Dict[str, Any]]],
119	]:
120	def wrapper(	1✔
121	args, *kwargs
122	) -> Union[Response, Tuple[Any, Dict[str, Any]]]:
123	input_filter = cls()	1✔
124	if request.method not in input_filter.__methods:	1✔
125	return Response(status=405, response="Method Not Allowed")	×
126
127	data = request.json if request.is_json else request.args	1✔
128
129	try:	1✔
130	kwargs = kwargs or {}	1✔
131
132	input_filter._data = {data, kwargs}	1✔
133
134	validated_data = input_filter.validateData()	1✔
135
136	if input_filter._model_class is not None:	1✔
137	g.validated_data = input_filter.serialize()	1✔
138	else:
139	g.validated_data = validated_data	1✔
140
141	except ValidationError as e:	1✔
142	return Response(	1✔
143	status=400,
144	response=json.dumps(e.args[0]),
145	mimetype="application/json",
146	)
147
NEW 148	except Exception:	×
NEW 149	logging.getLogger(__name__).exception(	×
150	"An unexpected exception occurred while "
151	"validating input data.",
152	)
NEW 153	return Response(status=500)	×
154
155	return f(args, *kwargs)	1✔
156
157	return wrapper	1✔
158
159	return decorator	1✔
160
161	@final	1✔
162	def validateData(	1✔
163	self, data: Optional[Dict[str, Any]] = None
164	) -> Dict[str, Any]:
165	"""
166	Validates input data against defined field rules, including applying
167	filters, validators, custom logic steps, and fallback mechanisms. The
168	validation process also ensures the required fields are handled
169	appropriately and conditions are checked after processing.
170
171	Args:
172	data (Dict[str, Any]): A dictionary containing the input data to
173	be validated where keys represent field names and values
174	represent the corresponding data.
175
176	Returns:
177	Dict[str, Any]: A dictionary containing the validated data with
178	any modifications, default values, or processed values as
179	per the defined validation rules.
180
181	Raises:
182	Any errors raised during external API calls, validation, or
183	logical steps execution of the respective fields or conditions
184	will propagate without explicit handling here.
185	"""
186	data = data or self._data	1✔
187	errors = {}	1✔
188
189	for field_name, field_info in self._fields.items():	1✔
190	value = data.get(field_name)	1✔
191
192	required = field_info.required	1✔
193	default = field_info.default	1✔
194	fallback = field_info.fallback	1✔
195	filters = field_info.filters	1✔
196	validators = field_info.validators	1✔
197	steps = field_info.steps	1✔
198	external_api = field_info.external_api	1✔
199	copy = field_info.copy	1✔
200
201	try:	1✔
202	if copy:	1✔
203	value = self._validated_data.get(copy)	1✔
204
205	if external_api:	1✔
206	value = self._ExternalApiMixin__callExternalApi(	1✔
207	external_api, fallback, self._validated_data
208	)
209
210	value = self._FilterMixin__applyFilters(filters, value)	1✔
211	value = (	1✔
212	self._ValidationMixin__validateField(
213	validators, fallback, value
214	)
215	or value
216	)
217	value = (	1✔
218	self._FieldMixin__applySteps(steps, fallback, value)
219	or value
220	)
221	value = self._FieldMixin__checkForRequired(	1✔
222	field_name, required, default, fallback, value
223	)
224
225	self._validated_data[field_name] = value	1✔
226
227	except ValidationError as e:	1✔
228	errors[field_name] = str(e)	1✔
229
230	try:	1✔
231	self._ConditionMixin__checkConditions(self._validated_data)	1✔
232	except ValidationError as e:	1✔
233	errors["_condition"] = str(e)	1✔
234
235	if errors:	1✔
236	raise ValidationError(errors)	1✔
237
238	return self._validated_data	1✔

LeanderCS / flask-inputfilter / #253

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous