LeanderCS / flask-inputfilter / #50

Committed 14 Jan 2025 09:28PM UTC coverage: 96.916%. Remained the same

Build # #50

Build Type

Pull #11

coveralls-python

Committed by

LeanderCS

Commit Message

10 | Add mounting in workflow to transfer the coverage data correctly

Pull Request Pull Request #11: 10 | Add testing for all supportet python versions

Run Details

1037 of 1070 relevant lines covered (96.92%)

0.97 hits per line

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

77.69

/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) -> None:
        self.fields = {}
        self.conditions = []

    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.
        :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 _applyFilters(self, field_name: str, value: Any) -> Any:
        """
        Apply filters to the field value.
        """

        field = self.fields.get(field_name)

        if not field:
            return value

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

        return value

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

        field = self.fields.get(field_name)

        if not field:
            return

        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(url: str, validated_data: dict) -> str:
        """
        Ersetzt alle Platzhalter in der URL, die mit {{}} definiert sind,
        durch die entsprechenden Werte aus den Parametern.
        """

        return re.sub(
            r"{{(.*?)}}",
            lambda match: str(validated_data.get(match.group(1))),
            url,
        )

    @staticmethod
    def __replacePlaceholdersInParams(
        params: dict, validated_data: dict
    ) -> dict:
        """
        Replace all placeholders in params with the
        corresponding values from validated_data.
        """
        replaced_params = {}
        for key, value in params.items():
            if isinstance(value, str):
                replaced_value = re.sub(
                    r"{{(.*?)}}",
                    lambda match: str(validated_data.get(match.group(1), "")),
                    value,
                )
                replaced_params[key] = replaced_value
            else:
                replaced_params[key] = value
        return replaced_params

    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 ValidationError:
                    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 == "GET":
                    data = request.args

                elif request.method in ["POST", "PUT", "DELETE"]:
                    if not request.is_json:
                        data = request.args

                    else:
                        data = request.json

                else:
                    return Response(
                        status=415, response="Unsupported method Type"
                    )

                inputFilter = cls()

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

LeanderCS / flask-inputfilter / #50

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