LeanderCS / flask-inputfilter / #53

Committed 15 Jan 2025 02:59PM UTC coverage: 99.052% (+2.1%) from 96.916%

Build # #53

Build Type

Pull #14

coveralls-python

Committed by

LeanderCS

Commit Message

12 | Increase test coverage for InputFilter

Pull Request Pull Request #14: 12 | New functionality for global filters and validators

Run Details

131 of 132 new or added lines in 31 files covered. (99.24%)

2 existing lines in 1 file now uncovered.

1045 of 1055 relevant lines covered (99.05%)

0.99 hits per line

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

96.55

/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,
        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 external_api: Configuration for an external API call.
        """

        self.fields[name] = {
            "required": required,
            "default": default,
            "fallback": fallback,
            "filters": filters or [],
            "validators": validators 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.
        """

        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, value: Any) -> None:
        """
        Validate the field value.
        """

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

        field = self.fields.get(field_name)

        for validator in field["validators"]:
            validator.validate(value)

    def _callExternalApi(
        self, config: ExternalApiConfig, 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

        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

    @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()}

    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)

            # Apply filters
            value = self._applyFilters(field_name, value)

            # Check for required field
            if value is None:
                if (
                    field_info.get("required")
                    and field_info.get("external_api") is None
                ):
                    if field_info.get("fallback") is None:
                        raise ValidationError(
                            f"Field '{field_name}' is required."
                        )

                    value = field_info.get("fallback")

                if field_info.get("default") is not None:
                    value = field_info.get("default")

            # Validate field
            if value is not None:
                try:
                    self._validateField(field_name, value)
                except ValidationError:
                    if field_info.get("fallback") is not None:
                        value = field_info.get("fallback")
                    else:
                        raise

            # External API call
            if field_info.get("external_api"):
                external_api_config = field_info.get("external_api")

                try:
                    value = self._callExternalApi(
                        external_api_config, validated_data
                    )

                except Exception:
                    if field_info.get("fallback") is None:
                        raise ValidationError(
                            f"External API call failed for field "
                            f"'{field_name}'."
                        )

                    value = field_info.get("fallback")

                if value is None:
                    if field_info.get("required"):
                        if field_info.get("fallback") is None:
                            raise ValidationError(
                                f"Field '{field_name}' is required."
                            )

                        value = field_info.get("fallback")

                    if field_info.get("default") is not None:
                        value = field_info.get("default")

            validated_data[field_name] = value

        # Check conditions
        for condition in self.conditions:
            if not condition.check(validated_data):
                raise ValidationError(f"Condition '{condition}' not met.")

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

LeanderCS / flask-inputfilter / #53

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