#122

Committed 29 Jan 2025 09:17PM UTC coverage: 98.657%. Remained the same

Build # #122

Build Type

push

coveralls-python

Committed by

LeanderCS

Commit Message

Merge pull request #28 from LeanderCS/27

27 | Optimize performance

Run Details

47 of 47 new or added lines in 9 files covered. (100.0%)

1 existing line in 1 file now uncovered.

1322 of 1340 relevant lines covered (98.66%)

0.99 hits per line

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

99.32

/flask_inputfilter/InputFilter.py

import re
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

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

from flask_inputfilter.Condition.BaseCondition import BaseCondition
from flask_inputfilter.Exception import ValidationError
from flask_inputfilter.Filter import BaseFilter
from flask_inputfilter.Model import ExternalApiConfig
from flask_inputfilter.Validator import BaseValidator


class InputFilter:
    """
    Base class for input filters.
    """

    def __init__(self, methods: Optional[List[str]] = None) -> None:
        self.methods = methods or ["GET", "POST", "PATCH", "PUT", "DELETE"]
        self.fields = {}
        self.conditions = []
        self.global_filters = []
        self.global_validators = []

    @final
    def add(
        self,
        name: str,
        required: bool = False,
        default: Any = None,
        fallback: Any = None,
        filters: Optional[List[BaseFilter]] = None,
        validators: Optional[List[BaseValidator]] = None,
        steps: Optional[List[Union[BaseFilter, BaseValidator]]] = None,
        external_api: Optional[ExternalApiConfig] = None,
        copy: Optional[str] = None,
    ) -> None:
        """
        Add the field to the input filter.

        :param name: The name of the field.
        :param required: Whether the field is required.
        :param default: The default value of the field.
        :param fallback: The fallback value of the field, if validations fails
        or field None, although it is required .
        :param filters: The filters to apply to the field value.
        :param validators: The validators to apply to the field value.
        :param steps: Allows to apply multiple filters and validators
        in a specific order.
        :param external_api: Configuration for an external API call.
        :param copy: The name of the field to copy the value from.
        """

        self.fields[name] = {
            "required": required,
            "default": default,
            "fallback": fallback,
            "filters": filters or [],
            "validators": validators or [],
            "steps": steps or [],
            "external_api": external_api,
            "copy": copy,
        }

    @final
    def addCondition(self, condition: BaseCondition) -> None:
        """
        Add a condition to the input filter.
        """
        self.conditions.append(condition)

    @final
    def addGlobalFilter(self, filter_: BaseFilter) -> None:
        """
        Add a global filter to be applied to all fields.
        """
        self.global_filters.append(filter_)

    @final
    def addGlobalValidator(self, validator: BaseValidator) -> None:
        """
        Add a global validator to be applied to all fields.
        """
        self.global_validators.append(validator)

    @final
    def __applyFilters(self, filters: List[BaseFilter], value: Any) -> Any:
        """
        Apply filters to the field value.
        """

        if value is None:
            return value

        for filter_ in self.global_filters + filters:
            value = filter_.apply(value)

        return value

    @final
    def __validateField(
        self, validators: List[BaseValidator], fallback: Any, value: Any
    ) -> None:
        """
        Validate the field value.
        """

        if value is None:
            return

        try:
            for validator in self.global_validators + validators:
                validator.validate(value)
        except ValidationError:
            if fallback is None:
                raise

            return fallback

    @final
    def __applySteps(
        self,
        steps: List[Union[BaseFilter, BaseValidator]],
        fallback: Any,
        value: Any,
    ) -> Any:
        """
        Apply multiple filters and validators in a specific order.
        """

        if value is None:
            return

        try:
            for step in steps:
                if isinstance(step, BaseFilter):
                    value = step.apply(value)
                elif isinstance(step, BaseValidator):
                    step.validate(value)
        except ValidationError:
            if fallback is None:
                raise
            return fallback

        return value

    @final
    def __callExternalApi(
        self, config: ExternalApiConfig, fallback: Any, validated_data: dict
    ) -> Optional[Any]:
        """
        Führt den API-Aufruf durch und gibt den Wert zurück,
        der im Antwortkörper zu finden ist.
        """

        requestData = {
            "headers": {},
            "params": {},
        }

        if config.api_key:
            requestData["headers"]["Authorization"] = (
                f"Bearer " f"{config.api_key}"
            )

        if config.headers:
            requestData["headers"].update(config.headers)

        if config.params:
            requestData["params"] = self.__replacePlaceholdersInParams(
                config.params, validated_data
            )

        requestData["url"] = self.__replacePlaceholders(
            config.url, validated_data
        )
        requestData["method"] = config.method

        try:
            response = requests.request(**requestData)

            if response.status_code != 200:
                raise ValidationError(
                    f"External API call failed with "
                    f"status code {response.status_code}"
                )

            result = response.json()

            data_key = config.data_key
            if data_key:
                return result.get(data_key)

            return result
        except Exception:
            if fallback is None:
                raise ValidationError(
                    f"External API call failed for field "
                    f"'{config.data_key}'."
                )

            return fallback

    @staticmethod
    @final
    def __replacePlaceholders(value: str, validated_data: dict) -> str:
        """
        Replace all placeholders, marked with '{{ }}' in value
        with the corresponding values from validated_data.
        """

        return re.sub(
            r"{{(.*?)}}",
            lambda match: str(validated_data.get(match.group(1))),
            value,
        )

    @final
    def __replacePlaceholdersInParams(
        self, params: dict, validated_data: dict
    ) -> dict:
        """
        Replace all placeholders in params with the
        corresponding values from validated_data.
        """
        return {
            key: self.__replacePlaceholders(value, validated_data)
            if isinstance(value, str)
            else value
            for key, value in params.items()
        }

    @staticmethod
    @final
    def __checkForRequired(
        field_name: str,
        required: bool,
        default: Any,
        fallback: Any,
        value: Any,
    ) -> Any:
        """
        Determine the value of the field, considering the required and
        fallback attributes.

        If the field is not required and no value is provided, the default
        value is returned.
        If the field is required and no value is provided, the fallback
        value is returned.
        If no of the above conditions are met, a ValidationError is raised.
        """

        if value is not None:
            return value

        if not required:
            return default

        if fallback is not None:
            return fallback

        raise ValidationError(f"Field '{field_name}' is required.")

    def __checkConditions(self, validated_data: dict) -> None:
        for condition in self.conditions:
            if not condition.check(validated_data):
                raise ValidationError(f"Condition '{condition}' not met.")

    @final
    def validateData(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Validate the input data, considering both request data and
        URL parameters (kwargs).
        """

        validated_data = {}

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

            if copy:
                value = validated_data.get(copy)

            if external_api:
                value = self.__callExternalApi(
                    external_api, fallback, validated_data
                )

            value = self.__applyFilters(filters, value)

            value = self.__validateField(validators, fallback, value) or value

            value = self.__applySteps(steps, fallback, value) or value

            value = self.__checkForRequired(
                field_name, required, default, fallback, value
            )

            validated_data[field_name] = value

        self.__checkConditions(validated_data)

        return validated_data

    @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:
                    g.validated_data = input_filter.validateData(data)

                except ValidationError as e:
                    return Response(status=400, response=str(e))

                return f(*args, **kwargs)

            return wrapper

        return decorator

1	import re	1✔
2	from typing import Any, Callable, Dict, List, Optional, Tuple, Union	1✔
3
4	import requests	1✔
5	from flask import Response, g, request	1✔
6	from typing_extensions import final	1✔
7
8	from flask_inputfilter.Condition.BaseCondition import BaseCondition	1✔
9	from flask_inputfilter.Exception import ValidationError	1✔
10	from flask_inputfilter.Filter import BaseFilter	1✔
11	from flask_inputfilter.Model import ExternalApiConfig	1✔
12	from flask_inputfilter.Validator import BaseValidator	1✔
13
14
15	class InputFilter:	1✔
16	"""
17	Base class for input filters.
18	"""
19
20	def __init__(self, methods: Optional[List[str]] = None) -> None:	1✔
21	self.methods = methods or ["GET", "POST", "PATCH", "PUT", "DELETE"]	1✔
22	self.fields = {}	1✔
23	self.conditions = []	1✔
24	self.global_filters = []	1✔
25	self.global_validators = []	1✔
26
27	@final	1✔
28	def add(	1✔
29	self,
30	name: str,
31	required: bool = False,
32	default: Any = None,
33	fallback: Any = None,
34	filters: Optional[List[BaseFilter]] = None,
35	validators: Optional[List[BaseValidator]] = None,
36	steps: Optional[List[Union[BaseFilter, BaseValidator]]] = None,
37	external_api: Optional[ExternalApiConfig] = None,
38	copy: Optional[str] = None,
39	) -> None:
40	"""
41	Add the field to the input filter.
42
43	:param name: The name of the field.
44	:param required: Whether the field is required.
45	:param default: The default value of the field.
46	:param fallback: The fallback value of the field, if validations fails
47	or field None, although it is required .
48	:param filters: The filters to apply to the field value.
49	:param validators: The validators to apply to the field value.
50	:param steps: Allows to apply multiple filters and validators
51	in a specific order.
52	:param external_api: Configuration for an external API call.
53	:param copy: The name of the field to copy the value from.
54	"""
55
56	self.fields[name] = {	1✔
57	"required": required,
58	"default": default,
59	"fallback": fallback,
60	"filters": filters or [],
61	"validators": validators or [],
62	"steps": steps or [],
63	"external_api": external_api,
64	"copy": copy,
65	}
66
67	@final	1✔
68	def addCondition(self, condition: BaseCondition) -> None:	1✔
69	"""
70	Add a condition to the input filter.
71	"""
72	self.conditions.append(condition)	1✔
73
74	@final	1✔
75	def addGlobalFilter(self, filter_: BaseFilter) -> None:	1✔
76	"""
77	Add a global filter to be applied to all fields.
78	"""
79	self.global_filters.append(filter_)	1✔
80
81	@final	1✔
82	def addGlobalValidator(self, validator: BaseValidator) -> None:	1✔
83	"""
84	Add a global validator to be applied to all fields.
85	"""
86	self.global_validators.append(validator)	1✔
87
88	@final	1✔
89	def __applyFilters(self, filters: List[BaseFilter], value: Any) -> Any:	1✔
90	"""
91	Apply filters to the field value.
92	"""
93
94	if value is None:	1✔
95	return value	1✔
96
97	for filter_ in self.global_filters + filters:	1✔
98	value = filter_.apply(value)	1✔
99
100	return value	1✔
101
102	@final	1✔
103	def __validateField(	1✔
104	self, validators: List[BaseValidator], fallback: Any, value: Any
105	) -> None:
106	"""
107	Validate the field value.
108	"""
109
110	if value is None:	1✔
111	return	1✔
112
113	try:	1✔
114	for validator in self.global_validators + validators:	1✔
115	validator.validate(value)	1✔
116	except ValidationError:	1✔
117	if fallback is None:	1✔
118	raise	1✔
119
120	return fallback	1✔
121
122	@final	1✔
123	def __applySteps(	1✔
124	self,
125	steps: List[Union[BaseFilter, BaseValidator]],
126	fallback: Any,
127	value: Any,
128	) -> Any:
129	"""
130	Apply multiple filters and validators in a specific order.
131	"""
132
133	if value is None:	1✔
134	return	1✔
135
136	try:	1✔
137	for step in steps:	1✔
138	if isinstance(step, BaseFilter):	1✔
139	value = step.apply(value)	1✔
140	elif isinstance(step, BaseValidator):	1✔
141	step.validate(value)	1✔
142	except ValidationError:	1✔
143	if fallback is None:	1✔
144	raise	1✔
145	return fallback	1✔
146
147	return value	1✔
148
149	@final	1✔
150	def __callExternalApi(	1✔
151	self, config: ExternalApiConfig, fallback: Any, validated_data: dict
152	) -> Optional[Any]:
153	"""
154	Führt den API-Aufruf durch und gibt den Wert zurück,
155	der im Antwortkörper zu finden ist.
156	"""
157
158	requestData = {	1✔
159	"headers": {},
160	"params": {},
161	}
162
163	if config.api_key:	1✔
164	requestData["headers"]["Authorization"] = (	1✔
165	f"Bearer " f"{config.api_key}"
166	)
167
168	if config.headers:	1✔
169	requestData["headers"].update(config.headers)	1✔
170
171	if config.params:	1✔
172	requestData["params"] = self.__replacePlaceholdersInParams(	1✔
173	config.params, validated_data
174	)
175
176	requestData["url"] = self.__replacePlaceholders(	1✔
177	config.url, validated_data
178	)
179	requestData["method"] = config.method	1✔
180
181	try:	1✔
182	response = requests.request(**requestData)	1✔
183
184	if response.status_code != 200:	1✔
185	raise ValidationError(	1✔
186	f"External API call failed with "
187	f"status code {response.status_code}"
188	)
189
190	result = response.json()	1✔
191
192	data_key = config.data_key	1✔
193	if data_key:	1✔
194	return result.get(data_key)	1✔
195
UNCOV 196	return result	×
197	except Exception:	1✔
198	if fallback is None:	1✔
199	raise ValidationError(	1✔
200	f"External API call failed for field "
201	f"'{config.data_key}'."
202	)
203
204	return fallback	1✔
205
206	@staticmethod	1✔
207	@final	1✔
208	def __replacePlaceholders(value: str, validated_data: dict) -> str:	1✔
209	"""
210	Replace all placeholders, marked with '{{ }}' in value
211	with the corresponding values from validated_data.
212	"""
213
214	return re.sub(	1✔
215	r"{{(.*?)}}",
216	lambda match: str(validated_data.get(match.group(1))),
217	value,
218	)
219
220	@final	1✔
221	def __replacePlaceholdersInParams(	1✔
222	self, params: dict, validated_data: dict
223	) -> dict:
224	"""
225	Replace all placeholders in params with the
226	corresponding values from validated_data.
227	"""
228	return {	1✔
229	key: self.__replacePlaceholders(value, validated_data)
230	if isinstance(value, str)
231	else value
232	for key, value in params.items()
233	}
234
235	@staticmethod	1✔
236	@final	1✔
237	def __checkForRequired(	1✔
238	field_name: str,
239	required: bool,
240	default: Any,
241	fallback: Any,
242	value: Any,
243	) -> Any:
244	"""
245	Determine the value of the field, considering the required and
246	fallback attributes.
247
248	If the field is not required and no value is provided, the default
249	value is returned.
250	If the field is required and no value is provided, the fallback
251	value is returned.
252	If no of the above conditions are met, a ValidationError is raised.
253	"""
254
255	if value is not None:	1✔
256	return value	1✔
257
258	if not required:	1✔
259	return default	1✔
260
261	if fallback is not None:	1✔
262	return fallback	1✔
263
264	raise ValidationError(f"Field '{field_name}' is required.")	1✔
265
266	def __checkConditions(self, validated_data: dict) -> None:	1✔
267	for condition in self.conditions:	1✔
268	if not condition.check(validated_data):	1✔
269	raise ValidationError(f"Condition '{condition}' not met.")	1✔
270
271	@final	1✔
272	def validateData(self, data: Dict[str, Any]) -> Dict[str, Any]:	1✔
273	"""
274	Validate the input data, considering both request data and
275	URL parameters (kwargs).
276	"""
277
278	validated_data = {}	1✔
279
280	for field_name, field_info in self.fields.items():	1✔
281	value = data.get(field_name)	1✔
282
283	required = field_info["required"]	1✔
284	default = field_info["default"]	1✔
285	fallback = field_info["fallback"]	1✔
286	filters = field_info["filters"]	1✔
287	validators = field_info["validators"]	1✔
288	steps = field_info["steps"]	1✔
289	external_api = field_info["external_api"]	1✔
290	copy = field_info["copy"]	1✔
291
292	if copy:	1✔
293	value = validated_data.get(copy)	1✔
294
295	if external_api:	1✔
296	value = self.__callExternalApi(	1✔
297	external_api, fallback, validated_data
298	)
299
300	value = self.__applyFilters(filters, value)	1✔
301
302	value = self.__validateField(validators, fallback, value) or value	1✔
303
304	value = self.__applySteps(steps, fallback, value) or value	1✔
305
306	value = self.__checkForRequired(	1✔
307	field_name, required, default, fallback, value
308	)
309
310	validated_data[field_name] = value	1✔
311
312	self.__checkConditions(validated_data)	1✔
313
314	return validated_data	1✔
315
316	@classmethod	1✔
317	@final	1✔
318	def validate(	1✔
319	cls,
320	) -> Callable[
321	[Any],
322	Callable[
323	[Tuple[Any, ...], Dict[str, Any]],
324	Union[Response, Tuple[Any, Dict[str, Any]]],
325	],
326	]:
327	"""
328	Decorator for validating input data in routes.
329	"""
330
331	def decorator(	1✔
332	f,
333	) -> Callable[
334	[Tuple[Any, ...], Dict[str, Any]],
335	Union[Response, Tuple[Any, Dict[str, Any]]],
336	]:
337	def wrapper(	1✔
338	args, *kwargs
339	) -> Union[Response, Tuple[Any, Dict[str, Any]]]:
340	input_filter = cls()	1✔
341	if request.method not in input_filter.methods:	1✔
342	return Response(status=405, response="Method Not Allowed")	1✔
343
344	data = request.json if request.is_json else request.args	1✔
345
346	try:	1✔
347	g.validated_data = input_filter.validateData(data)	1✔
348
349	except ValidationError as e:	1✔
350	return Response(status=400, response=str(e))	1✔
351
352	return f(args, *kwargs)	1✔
353
354	return wrapper	1✔
355
356	return decorator	1✔

LeanderCS / flask-inputfilter / #122

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