LeanderCS / flask-inputfilter / #60

Committed 15 Jan 2025 11:45PM UTC coverage: 99.337% (+0.3%) from 99.052%

Build # #60

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #19 from LeanderCS/13

13 | Refactor

Run Details

53 of 54 new or added lines in 5 files covered. (98.15%)

2 existing lines in 2 files now uncovered.

1049 of 1056 relevant lines covered (99.34%)

0.99 hits per line

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

99.15

/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:
        :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["filters"]:
            value = filter_.apply(value)

        return value

    def __validateField(self, field_name: str, field_info, 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["validators"]:
                validator.validate(value)
        except ValidationError:
            if field_info.get("fallback") is None:
                raise

            return field_info.get("fallback")

    def __callExternalApi(
        self, field_info, 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
            )

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

LeanderCS / flask-inputfilter / #60

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