LeanderCS / flask-inputfilter / #73

Committed 18 Jan 2025 10:58PM UTC coverage: 98.065% (-1.0%) from 99.052%

Build # #73

Build Type

push

coveralls-python

Committed by

LeanderCS

Commit Message

15 | Add caching to workflow

Run Details

1216 of 1240 relevant lines covered (98.06%)

0.98 hits per line

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

99.25

/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 .Condition.BaseCondition import BaseCondition
from .Exception import ValidationError
from .Filter.BaseFilter import BaseFilter
from .Model import ExternalApiConfig
from .Validator.BaseValidator 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 = []

    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,
    ) -> 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.
        """

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

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

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

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

    def __applyFilters(self, field_name: str, value: Any) -> Any:
        """
        Apply filters to the field value.
        """

        if value is None:
            return value

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

        field = self.fields.get(field_name)

        for filter_ in field.get("filters"):
            value = filter_.apply(value)

        return value

    def __validateField(
        self, field_name: str, field_info: Any, value: Any
    ) -> None:
        """
        Validate the field value.
        """

        if value is None:
            return

        try:
            for validator in self.global_validators:
                validator.validate(value)

            field = self.fields.get(field_name)

            for validator in field.get("validators"):
                validator.validate(value)
        except ValidationError:
            if field_info.get("fallback") is None:
                raise

            return field_info.get("fallback")

    def __applySteps(
        self, field_name: str, field_info: Any, value: Any
    ) -> Any:
        """
        Apply multiple filters and validators in a specific order.
        """

        if value is None:
            return

        field = self.fields.get(field_name)

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

        return value

    def __callExternalApi(
        self, field_info: 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.
        """

        config: ExternalApiConfig = field_info.get("external_api")

        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 field_info and field_info.get("fallback") is None:
                raise ValidationError(
                    f"External API call failed for field "
                    f"'{config.data_key}'."
                )

            return field_info.get("fallback")

    @staticmethod
    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,
        )

    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
    def __checkForRequired(
        field_name: str, field_info: dict, 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 field_info.get("required"):
            return field_info.get("default")

        if field_info.get("fallback") is not None:
            return field_info.get("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.")

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

        if kwargs is None:
            kwargs = {}

        validated_data = {}
        combined_data = {**data, **kwargs}

        for field_name, field_info in self.fields.items():
            value = combined_data.get(field_name)

            value = self.__applyFilters(field_name, value)

            value = (
                self.__validateField(field_name, field_info, value) or value
            )

            value = self.__applySteps(field_name, field_info, value) or value

            if field_info.get("external_api"):
                value = self.__callExternalApi(field_info, validated_data)

            value = self.__checkForRequired(field_name, field_info, value)

            validated_data[field_name] = value

        self.__checkConditions(validated_data)

        return validated_data

    @classmethod
    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]]]:
                if request.method not in cls().methods:
                    return Response(status=405, response="Method Not Allowed")

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

                try:
                    g.validated_data = cls().validateData(data, kwargs)

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

LeanderCS / flask-inputfilter / #73

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