LeanderCS / flask-inputfilter / #58

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

Build # #58

Build Type

Pull #19

coveralls-python

Committed by

LeanderCS

Commit Message

13 | Refactor

Pull Request Pull Request #19: 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 fallback attributes.
210
211	If the field is not required and no value is provided, the default value is returned.
212	If the field is required and no value is provided, the fallback value is returned.
213	If no of the above conditions are met, a ValidationError is raised.
214	"""
215
216	if value is not None:	1✔
217	return value	1✔
218
219	if not field_info.get("required"):	1✔
220	return field_info.get("default")	1✔
221
222	if field_info.get("fallback") is not None:	1✔
223	return field_info.get("fallback")	1✔
224
225	raise ValidationError(f"Field '{field_name}' is required.")	1✔
226
227	def __checkConditions(self, validated_data: dict) -> None:	1✔
228	for condition in self.conditions:	1✔
229	if not condition.check(validated_data):	1✔
230	raise ValidationError(f"Condition '{condition}' not met.")	1✔
231
232	def validateData(	1✔
233	self, data: Dict[str, Any], kwargs: Dict[str, Any] = None
234	) -> Dict[str, Any]:
235	"""
236	Validate the input data, considering both request data and
237	URL parameters (kwargs).
238	"""
239
240	if kwargs is None:	1✔
241	kwargs = {}	1✔
242
243	validated_data = {}	1✔
244	combined_data = {data, kwargs}	1✔
245
246	for field_name, field_info in self.fields.items():	1✔
247	value = combined_data.get(field_name)	1✔
248
249	value = self.__applyFilters(field_name, value)	1✔
250
251	value = (	1✔
252	self.__validateField(field_name, field_info, value) or value
253	)
254
255	if field_info.get("external_api"):	1✔
256	value = self.__callExternalApi(field_info, validated_data)	1✔
257
258	value = self.__checkForRequired(field_name, field_info, value)	1✔
259
260	validated_data[field_name] = value	1✔
261
262	self.__checkConditions(validated_data)	1✔
263
264	return validated_data	1✔
265
266	@classmethod	1✔
267	def validate(	1✔
268	cls,
269	) -> Callable[
270	[Any],
271	Callable[
272	[Tuple[Any, ...], Dict[str, Any]],
273	Union[Response, Tuple[Any, Dict[str, Any]]],
274	],
275	]:
276	"""
277	Decorator for validating input data in routes.
278	"""
279
280	def decorator(	1✔
281	f,
282	) -> Callable[
283	[Tuple[Any, ...], Dict[str, Any]],
284	Union[Response, Tuple[Any, Dict[str, Any]]],
285	]:
286	def wrapper(	1✔
287	args, *kwargs
288	) -> Union[Response, Tuple[Any, Dict[str, Any]]]:
289	if request.method not in cls().methods:	1✔
290	return Response(status=405, response="Method Not Allowed")	1✔
291
292	data = request.json if request.is_json else request.args	1✔
293
294	try:	1✔
295	g.validated_data = cls().validateData(data, kwargs)	1✔
296
297	except ValidationError as e:	1✔
298	return Response(status=400, response=str(e))	1✔
299
300	return f(args, *kwargs)	1✔
301
302	return wrapper	1✔
303
304	return decorator	1✔

LeanderCS / flask-inputfilter / #58

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