Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/validators/form_data.py: 41%

1import logging 

2import typing as t 

3 

4from jsonschema import ValidationError, draft4_format_checker 

5from starlette.datastructures import Headers, UploadFile 

6from starlette.formparsers import FormParser, MultiPartParser 

7from starlette.types import Scope 

8 

9from connexion.exceptions import BadRequestProblem, ExtraParameterProblem 

10from connexion.json_schema import Draft4RequestValidator, format_error_with_path 

11from connexion.uri_parsing import AbstractURIParser 

12from connexion.validators import AbstractRequestBodyValidator 

13 

14logger = logging.getLogger("connexion.validators.form_data") 

15 

16 

17class FormDataValidator(AbstractRequestBodyValidator): 

18 """Request body validator for form content types.""" 

19 

20 def __init__( 

21 self, 

22 *, 

23 schema: dict, 

24 required=False, 

25 nullable=False, 

26 encoding: str, 

27 strict_validation: bool, 

28 uri_parser: t.Optional[AbstractURIParser] = None, 

29 ) -> None: 

30 super().__init__( 

31 schema=schema, 

32 required=required, 

33 nullable=nullable, 

34 encoding=encoding, 

35 strict_validation=strict_validation, 

36 ) 

37 self._uri_parser = uri_parser 

38 

39 @property 

40 def _validator(self): 

41 return Draft4RequestValidator( 

42 self._schema, format_checker=draft4_format_checker 

43 ) 

44 

45 @property 

46 def _form_parser_cls(self): 

47 return FormParser 

48 

49 async def _parse(self, stream: t.AsyncGenerator[bytes, None], scope: Scope) -> dict: 

50 headers = Headers(scope=scope) 

51 form_parser = self._form_parser_cls(headers, stream) 

52 data = await form_parser.parse() 

53 

54 if self._uri_parser is not None: 

55 # Don't parse file_data 

56 form_data = {} 

57 file_data = {} 

58 for k, v in data.items(): 

59 if isinstance(v, str): 

60 form_data[k] = data.getlist(k) 

61 elif isinstance(v, UploadFile): 

62 # Replace files with empty strings for validation 

63 file_data[k] = "" 

64 

65 data = self._uri_parser.resolve_form(form_data) 

66 # Add the files again 

67 data.update(file_data) 

68 else: 

69 data = {k: data.getlist(k) for k in data} 

70 

71 return data 

72 

73 def _validate(self, data: dict) -> None: 

74 if self._strict_validation: 

75 self._validate_params_strictly(data) 

76 

77 try: 

78 self._validator.validate(data) 

79 except ValidationError as exception: 

80 error_path_msg = format_error_with_path(exception=exception) 

81 logger.error( 

82 f"Validation error: {exception.message}{error_path_msg}", 

83 extra={"validator": "body"}, 

84 ) 

85 raise BadRequestProblem(detail=f"{exception.message}{error_path_msg}") 

86 

87 def _validate_params_strictly(self, data: dict) -> None: 

88 form_params = data.keys() 

89 spec_params = self._schema.get("properties", {}).keys() 

90 errors = set(form_params).difference(set(spec_params)) 

91 if errors: 

92 raise ExtraParameterProblem(param_type="formData", extra_params=errors) 

93 

94 

95class MultiPartFormDataValidator(FormDataValidator): 

96 @property 

97 def _form_parser_cls(self): 

98 return MultiPartParser