LeanderCS / flask-inputfilter / #68

Committed 16 Jan 2025 11:08PM UTC coverage: 99.252% (-0.09%) from 99.337%

Build # #68

Build Type

Pull #20

coveralls-python

Committed by

LeanderCS

Commit Message

16 | Add functionality to filter and validate in a custom order

Pull Request Pull Request #20: 16 | Add functionality to filter and validate in a custom order

Coverage Stats

16 of 17 new or added lines in 1 file covered. (94.12%)

1062 of 1070 relevant lines covered (99.25%)

0.99 hits per line

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

98.47

/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.
        """

        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)

            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	field = self.fields.get(field_name)	1✔
130
131	try:	1✔
132	for step in field.get("steps"):	1✔
133	if isinstance(step, BaseFilter):	1✔
134	value = step.apply(value)	1✔
135	elif isinstance(step, BaseValidator):	1✔
136	step.validate(value)	1✔
137	except ValidationError:	1✔
138	if field_info.get("fallback") is None:	1✔
139	raise	1✔
NEW 140	return field_info.get("fallback")	×
141
142	return value	1✔
143
144	def __callExternalApi(	1✔
145	self, field_info: Any, validated_data: dict
146	) -> Optional[Any]:
147	"""
148	Führt den API-Aufruf durch und gibt den Wert zurück,
149	der im Antwortkörper zu finden ist.
150	"""
151
152	config: ExternalApiConfig = field_info.get("external_api")	1✔
153
154	requestData = {	1✔
155	"headers": {},
156	"params": {},
157	}
158
159	if config.api_key:	1✔
160	requestData["headers"]["Authorization"] = (	1✔
161	f"Bearer " f"{config.api_key}"
162	)
163
164	if config.headers:	1✔
165	requestData["headers"].update(config.headers)	1✔
166
167	if config.params:	1✔
168	requestData["params"] = self.__replacePlaceholdersInParams(	1✔
169	config.params, validated_data
170	)
171
172	requestData["url"] = self.__replacePlaceholders(	1✔
173	config.url, validated_data
174	)
175	requestData["method"] = config.method	1✔
176
177	try:	1✔
178	response = requests.request(**requestData)	1✔
179
180	if response.status_code != 200:	1✔
181	raise ValidationError(	1✔
182	f"External API call failed with "
183	f"status code {response.status_code}"
184	)
185
186	result = response.json()	1✔
187
188	data_key = config.data_key	1✔
189	if data_key:	1✔
190	return result.get(data_key)	1✔
191
192	return result	×
193	except Exception:	1✔
194	if field_info and field_info.get("fallback") is None:	1✔
195	raise ValidationError(	1✔
196	f"External API call failed for field "
197	f"'{config.data_key}'."
198	)
199
200	return field_info.get("fallback")	1✔
201
202	@staticmethod	1✔
203	def __replacePlaceholders(value: str, validated_data: dict) -> str:	1✔
204	"""
205	Replace all placeholders, marked with '{{ }}' in value
206	with the corresponding values from validated_data.
207	"""
208
209	return re.sub(	1✔
210	r"{{(.*?)}}",
211	lambda match: str(validated_data.get(match.group(1))),
212	value,
213	)
214
215	def __replacePlaceholdersInParams(	1✔
216	self, params: dict, validated_data: dict
217	) -> dict:
218	"""
219	Replace all placeholders in params with the
220	corresponding values from validated_data.
221	"""
222	return {	1✔
223	key: self.__replacePlaceholders(value, validated_data)
224	if isinstance(value, str)
225	else value
226	for key, value in params.items()
227	}
228
229	@staticmethod	1✔
230	def __checkForRequired(	1✔
231	field_name: str, field_info: dict, value: Any
232	) -> Any:
233	"""
234	Determine the value of the field, considering the required and
235	fallback attributes.
236
237	If the field is not required and no value is provided, the default
238	value is returned.
239	If the field is required and no value is provided, the fallback
240	value is returned.
241	If no of the above conditions are met, a ValidationError is raised.
242	"""
243
244	if value is not None:	1✔
245	return value	1✔
246
247	if not field_info.get("required"):	1✔
248	return field_info.get("default")	1✔
249
250	if field_info.get("fallback") is not None:	1✔
251	return field_info.get("fallback")	1✔
252
253	raise ValidationError(f"Field '{field_name}' is required.")	1✔
254
255	def __checkConditions(self, validated_data: dict) -> None:	1✔
256	for condition in self.conditions:	1✔
257	if not condition.check(validated_data):	1✔
258	raise ValidationError(f"Condition '{condition}' not met.")	1✔
259
260	def validateData(	1✔
261	self, data: Dict[str, Any], kwargs: Dict[str, Any] = None
262	) -> Dict[str, Any]:
263	"""
264	Validate the input data, considering both request data and
265	URL parameters (kwargs).
266	"""
267
268	if kwargs is None:	1✔
269	kwargs = {}	1✔
270
271	validated_data = {}	1✔
272	combined_data = {data, kwargs}	1✔
273
274	for field_name, field_info in self.fields.items():	1✔
275	value = combined_data.get(field_name)	1✔
276
277	value = self.__applyFilters(field_name, value)	1✔
278
279	value = (	1✔
280	self.__validateField(field_name, field_info, value) or value
281	)
282
283	value = self.__applySteps(field_name, field_info, value)	1✔
284
285	if field_info.get("external_api"):	1✔
286	value = self.__callExternalApi(field_info, validated_data)	1✔
287
288	value = self.__checkForRequired(field_name, field_info, value)	1✔
289
290	validated_data[field_name] = value	1✔
291
292	self.__checkConditions(validated_data)	1✔
293
294	return validated_data	1✔
295
296	@classmethod	1✔
297	def validate(	1✔
298	cls,
299	) -> Callable[
300	[Any],
301	Callable[
302	[Tuple[Any, ...], Dict[str, Any]],
303	Union[Response, Tuple[Any, Dict[str, Any]]],
304	],
305	]:
306	"""
307	Decorator for validating input data in routes.
308	"""
309
310	def decorator(	1✔
311	f,
312	) -> Callable[
313	[Tuple[Any, ...], Dict[str, Any]],
314	Union[Response, Tuple[Any, Dict[str, Any]]],
315	]:
316	def wrapper(	1✔
317	args, *kwargs
318	) -> Union[Response, Tuple[Any, Dict[str, Any]]]:
319	if request.method not in cls().methods:	1✔
320	return Response(status=405, response="Method Not Allowed")	1✔
321
322	data = request.json if request.is_json else request.args	1✔
323
324	try:	1✔
325	g.validated_data = cls().validateData(data, kwargs)	1✔
326
327	except ValidationError as e:	1✔
328	return Response(status=400, response=str(e))	1✔
329
330	return f(args, *kwargs)	1✔
331
332	return wrapper	1✔
333
334	return decorator	1✔

LeanderCS / flask-inputfilter / #68

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