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

1import json 

2import logging 

3import typing as t 

4 

5import jsonschema 

6from jsonschema import Draft4Validator, ValidationError, draft4_format_checker 

7from starlette.types import Scope 

8 

9from connexion.exceptions import BadRequestProblem, NonConformingResponseBody 

10from connexion.json_schema import ( 

11 Draft4RequestValidator, 

12 Draft4ResponseValidator, 

13 format_error_with_path, 

14) 

15from connexion.validators import ( 

16 AbstractRequestBodyValidator, 

17 AbstractResponseBodyValidator, 

18) 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23class JSONRequestBodyValidator(AbstractRequestBodyValidator): 

24 """Request body validator for json content types.""" 

25 

26 def __init__( 

27 self, 

28 *, 

29 schema: dict, 

30 required=False, 

31 nullable=False, 

32 encoding: str, 

33 strict_validation: bool, 

34 **kwargs, 

35 ) -> None: 

36 super().__init__( 

37 schema=schema, 

38 required=required, 

39 nullable=nullable, 

40 encoding=encoding, 

41 strict_validation=strict_validation, 

42 ) 

43 

44 @property 

45 def _validator(self): 

46 return Draft4RequestValidator( 

47 self._schema, format_checker=draft4_format_checker 

48 ) 

49 

50 async def _parse( 

51 self, stream: t.AsyncGenerator[bytes, None], scope: Scope 

52 ) -> t.Any: 

53 bytes_body = b"".join([message async for message in stream]) 

54 body = bytes_body.decode(self._encoding) 

55 

56 if not body: 

57 return None 

58 

59 try: 

60 return json.loads(body) 

61 except json.decoder.JSONDecodeError as e: 

62 raise BadRequestProblem(detail=str(e)) 

63 

64 def _validate(self, body: dict) -> None: 

65 try: 

66 return self._validator.validate(body) 

67 except ValidationError as exception: 

68 error_path_msg = format_error_with_path(exception=exception) 

69 logger.error( 

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

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

72 ) 

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

74 

75 

76class DefaultsJSONRequestBodyValidator(JSONRequestBodyValidator): 

77 """Request body validator for json content types which fills in default values. This Validator 

78 intercepts the body, makes changes to it, and replays it for the next ASGI application.""" 

79 

80 MUTABLE_VALIDATION = True 

81 """This validator might mutate to the body.""" 

82 

83 @property 

84 def _validator(self): 

85 validator_cls = self.extend_with_set_default(Draft4RequestValidator) 

86 return validator_cls(self._schema, format_checker=draft4_format_checker) 

87 

88 # via https://python-jsonschema.readthedocs.io/ 

89 @staticmethod 

90 def extend_with_set_default(validator_class): 

91 validate_properties = validator_class.VALIDATORS["properties"] 

92 

93 def set_defaults(validator, properties, instance, schema): 

94 for property, subschema in properties.items(): 

95 if "default" in subschema: 

96 instance.setdefault(property, subschema["default"]) 

97 

98 yield from validate_properties(validator, properties, instance, schema) 

99 

100 return jsonschema.validators.extend( 

101 validator_class, {"properties": set_defaults} 

102 ) 

103 

104 

105class JSONResponseBodyValidator(AbstractResponseBodyValidator): 

106 """Response body validator for json content types.""" 

107 

108 @property 

109 def validator(self) -> Draft4Validator: 

110 return Draft4ResponseValidator( 

111 self._schema, format_checker=draft4_format_checker 

112 ) 

113 

114 def _parse(self, stream: t.Generator[bytes, None, None]) -> t.Any: 

115 body = b"".join(stream).decode(self._encoding) 

116 

117 if not body: 

118 return None 

119 

120 try: 

121 return json.loads(body) 

122 except json.decoder.JSONDecodeError as e: 

123 raise NonConformingResponseBody(str(e)) 

124 

125 def _validate(self, body: dict): 

126 try: 

127 self.validator.validate(body) 

128 except ValidationError as exception: 

129 error_path_msg = format_error_with_path(exception=exception) 

130 logger.error( 

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

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

133 ) 

134 raise NonConformingResponseBody( 

135 detail=f"Response body does not conform to specification. {exception.message}{error_path_msg}" 

136 ) 

137 

138 

139class TextResponseBodyValidator(JSONResponseBodyValidator): 

140 def _parse(self, stream: t.Generator[bytes, None, None]) -> str: # type: ignore 

141 body = b"".join(stream).decode(self._encoding) 

142 

143 try: 

144 return json.loads(body) 

145 except json.decoder.JSONDecodeError: 

146 return body