LeanderCS / flask-inputfilter / #42

Committed 14 Jan 2025 12:35PM UTC coverage: 96.916% (+0.2%) from 96.729%

Build # #42

Build Type

push

coveralls-python

Committed by

LeanderCS

Commit Message

8 | Extend docs

Run Details

40 of 40 new or added lines in 4 files covered. (100.0%)

16 existing lines in 1 file now uncovered.

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 = {}

        if config.api_key:
            requestData["headers"][
                "Authorization"
            ] = f"Bearer {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 status code "
                f"{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
98	if config.api_key:
99	requestData["headers"][
100	"Authorization"
101	] = f"Bearer {config.api_key}"	1✔
102		1✔
103	if config.headers:
104	requestData["headers"].update(config.headers)
105		1✔
106	if config.params:	1✔
107	requestData["params"] = self.__replacePlaceholdersInParams(
108	config.params, validated_data	1✔
109	)	1✔
110
111	requestData["url"] = self.__replacePlaceholders(
112	config.url, validated_data	1✔
113	)
114	requestData["method"] = config.method	1✔
115
116	response = requests.request(**requestData)	1✔
117
118	if response.status_code != 200:	1✔
119	raise ValidationError(	1✔
120	f"External API call failed with status code "
121	f"{response.status_code}"
122	)	1✔
123
124	result = response.json()	1✔
125		1✔
126	data_key = config.data_key	1✔
127	if data_key:
UNCOV 128	return result.get(data_key)	×
129
130	return result	1✔
131		1✔
132	@staticmethod
133	def __replacePlaceholders(url: str, validated_data: dict) -> str:
134	"""
135	Ersetzt alle Platzhalter in der URL, die mit {{}} definiert sind,
136	durch die entsprechenden Werte aus den Parametern.
137	"""	1✔
138
139	return re.sub(
140	r"{{(.*?)}}",
141	lambda match: str(validated_data.get(match.group(1))),
142	url,
143	)	1✔
144		1✔
145	@staticmethod
146	def __replacePlaceholdersInParams(
147	params: dict, validated_data: dict
148	) -> dict:
149	"""
150	Replace all placeholders in params with the
151	corresponding values from validated_data.	1✔
152	"""	1✔
153	replaced_params = {}	1✔
154	for key, value in params.items():	1✔
155	if isinstance(value, str):
156	replaced_value = re.sub(
157	r"{{(.*?)}}",
158	lambda match: str(validated_data.get(match.group(1), "")),
159	value,	1✔
160	)
UNCOV 161	replaced_params[key] = replaced_value	×
162	else:	1✔
163	replaced_params[key] = value
164	return replaced_params	1✔
165
166	def validateData(
167	self, data: Dict[str, Any], kwargs: Dict[str, Any] = None
168	) -> Dict[str, Any]:
169	"""
170	Validate the input data, considering both request data and
171	URL parameters (kwargs).
172	"""	1✔
173		1✔
174	if kwargs is None:
175	kwargs = {}	1✔
176		1✔
177	validated_data = {}
178	combined_data = {data, kwargs}	1✔
179		1✔
180	for field_name, field_info in self.fields.items():
181	value = combined_data.get(field_name)
182		1✔
183	# Apply filters
184	value = self._applyFilters(field_name, value)
185		1✔
186	# Check for required field	1✔
187	if value is None:
188	if (
189	field_info.get("required")
190	and field_info.get("external_api") is None	1✔
191	):	1✔
192	if field_info.get("fallback") is None:
193	raise ValidationError(
194	f"Field '{field_name}' is required."
195	)	1✔
196
197	value = field_info.get("fallback")	1✔
198		1✔
199	if field_info.get("default") is not None:
200	value = field_info.get("default")
201		1✔
202	# Validate field	1✔
203	if value is not None:	1✔
204	try:	1✔
205	self._validateField(field_name, value)	1✔
206	except ValidationError:	1✔
207	if field_info.get("fallback") is not None:
208	value = field_info.get("fallback")	1✔
209	else:
210	raise
211		1✔
212	# External API call	1✔
213	if field_info.get("external_api"):
214	external_api_config = field_info.get("external_api")	1✔
215		1✔
216	try:
217	value = self._callExternalApi(
218	external_api_config, validated_data
219	)	1✔
220		1✔
221	except ValidationError:	1✔
222	if field_info.get("fallback") is None:
223	raise ValidationError(
224	f"External API call failed for field "
225	f"'{field_name}'."
226	)	1✔
227
228	value = field_info.get("fallback")	1✔
UNCOV 229		×
UNCOV 230	if value is None:	×
231	if field_info.get("required"):	×
232	if field_info.get("fallback") is None:
233	raise ValidationError(
234	f"Field '{field_name}' is required."
UNCOV 235	)	×
236
237	value = field_info.get("fallback")	×
UNCOV 238		×
239	if field_info.get("default") is not None:
240	value = field_info.get("default")	1✔
241
242	validated_data[field_name] = value
243		1✔
244	# Check conditions	1✔
245	for condition in self.conditions:	1✔
246	if not condition.check(validated_data):
247	raise ValidationError(f"Condition '{condition}' not met.")	1✔
248
249	return validated_data	1✔
250		1✔
251	@classmethod
252	def validate(
253	cls,
254	) -> Callable[
255	[Any],
256	Callable[
257	[Tuple[Any, ...], Dict[str, Any]],
258	Union[Response, Tuple[Any, Dict[str, Any]]],
259	],
260	]:
261	"""
262	Decorator for validating input data in routes.
UNCOV 263	"""	×
264
265	def decorator(
266	f,
267	) -> Callable[
268	[Tuple[Any, ...], Dict[str, Any]],
UNCOV 269	Union[Response, Tuple[Any, Dict[str, Any]]],	×
270	]:
271	def wrapper(
UNCOV 272	args, *kwargs	×
UNCOV 273	) -> Union[Response, Tuple[Any, Dict[str, Any]]]:	×
274	if request.method == "GET":
275	data = request.args	×
UNCOV 276		×
277	elif request.method in ["POST", "PUT", "DELETE"]:	×
278	if not request.is_json:
279	data = request.args
UNCOV 280		×
281	else:
282	data = request.json
UNCOV 283		×
284	else:
285	return Response(
286	status=415, response="Unsupported method Type"
UNCOV 287	)	×
288
289	inputFilter = cls()	×
UNCOV 290		×
291	try:
292	g.validated_data = inputFilter.validateData(data, kwargs)	×
UNCOV 293		×
294	except ValidationError as e:
295	return Response(status=400, response=str(e))	×
296
297	return f(args, *kwargs)	×
298
299	return wrapper	×
300
301	return decorator

LeanderCS / flask-inputfilter / #42

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