Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/decorators/response.py: 29%

1import abc 

2import collections.abc 

3import functools 

4import logging 

5import types 

6import typing as t 

7from enum import Enum 

8 

9from connexion.context import operation 

10from connexion.datastructures import NoContent 

11from connexion.exceptions import NonConformingResponseHeaders 

12from connexion.frameworks.abstract import Framework 

13from connexion.lifecycle import ConnexionResponse 

14from connexion.utils import is_json_mimetype 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class BaseResponseDecorator: 

20 def __init__(self, *, framework: t.Type[Framework], jsonifier): 

21 self.framework = framework 

22 self.jsonifier = jsonifier 

23 

24 @abc.abstractmethod 

25 def __call__(self, function: t.Callable) -> t.Callable: 

26 raise NotImplementedError 

27 

28 def build_framework_response(self, handler_response): 

29 data, status_code, headers = self._unpack_handler_response(handler_response) 

30 content_type = self._deduct_content_type(data, headers) 

31 if not self.framework.is_framework_response(data): 

32 data, status_code = self._prepare_body_and_status_code( 

33 data, status_code=status_code, mimetype=content_type 

34 ) 

35 return self.framework.build_response( 

36 data, content_type=content_type, status_code=status_code, headers=headers 

37 ) 

38 

39 @staticmethod 

40 def _deduct_content_type(data: t.Any, headers: dict) -> str: 

41 """Deduct the response content type from the returned data, headers and operation spec. 

42 

43 :param data: Response data 

44 :param headers: Headers returned by the handler. 

45 

46 :return: Deducted content type 

47 

48 :raises: NonConformingResponseHeaders if content type cannot be deducted. 

49 """ 

50 content_type = headers.get("Content-Type") 

51 

52 # TODO: don't default 

53 produces = list(set(operation.produces)) 

54 if data is not None and not produces: 

55 produces = ["application/json"] 

56 

57 if content_type: 

58 if content_type not in produces: 

59 raise NonConformingResponseHeaders( 

60 f"Returned content type ({content_type}) is not defined in operation spec " 

61 f"({operation.produces})." 

62 ) 

63 else: 

64 if not produces: 

65 # Produces can be empty/ for empty responses 

66 pass 

67 elif len(produces) == 1: 

68 content_type = produces[0] 

69 elif isinstance(data, str) and "text/plain" in produces: 

70 content_type = "text/plain" 

71 elif ( 

72 isinstance(data, bytes) 

73 or isinstance(data, (types.GeneratorType, collections.abc.Iterator)) 

74 ) and "application/octet-stream" in produces: 

75 content_type = "application/octet-stream" 

76 else: 

77 raise NonConformingResponseHeaders( 

78 "Multiple response content types are defined in the operation spec, but the " 

79 "handler response did not specify which one to return." 

80 ) 

81 

82 return content_type 

83 

84 def _prepare_body_and_status_code( 

85 self, data, *, status_code: int = None, mimetype: str 

86 ) -> tuple: 

87 if data is NoContent: 

88 data = None 

89 

90 if status_code is None: 

91 if data is None: 

92 status_code = 204 

93 else: 

94 status_code = 200 

95 

96 if data is not None: 

97 body = self._serialize_data(data, mimetype) 

98 else: 

99 body = data 

100 

101 return body, status_code 

102 

103 def _serialize_data(self, data: t.Any, mimetype: str) -> t.Any: 

104 if is_json_mimetype(mimetype): 

105 return self.jsonifier.dumps(data) 

106 return data 

107 

108 @staticmethod 

109 def _unpack_handler_response( 

110 handler_response: t.Union[str, bytes, dict, list, tuple] 

111 ) -> t.Tuple[t.Union[str, bytes, dict, list, None], t.Optional[int], dict]: 

112 """Unpack the handler response into data, status_code and headers. 

113 

114 :param handler_response: The response returned from the handler function if it was not a 

115 response class. 

116 

117 :return: A tuple of data, status_code and headers 

118 """ 

119 data, status_code, headers = None, None, {} 

120 

121 if not isinstance(handler_response, tuple): 

122 data = handler_response 

123 

124 elif len(handler_response) == 1: 

125 (data,) = handler_response 

126 

127 elif len(handler_response) == 2: 

128 data, status_code_or_headers = handler_response 

129 if isinstance(status_code_or_headers, int): 

130 status_code = status_code_or_headers 

131 elif isinstance(status_code_or_headers, Enum) and isinstance( 

132 status_code_or_headers.value, int 

133 ): 

134 status_code = status_code_or_headers.value 

135 else: 

136 headers = status_code_or_headers 

137 

138 elif len(handler_response) == 3: 

139 data, status_code, headers = handler_response 

140 

141 else: 

142 raise TypeError( 

143 "The view function did not return a valid response tuple." 

144 " The tuple must have the form (body), (body, status, headers)," 

145 " (body, status), or (body, headers)." 

146 ) 

147 

148 return data, status_code, headers 

149 

150 

151class SyncResponseDecorator(BaseResponseDecorator): 

152 def __call__(self, function: t.Callable) -> t.Callable: 

153 @functools.wraps(function) 

154 def wrapper(*args, **kwargs): 

155 """ 

156 This method converts a handler response to a framework response. 

157 The handler response can be a ConnexionResponse, a framework response, a tuple or an 

158 object. 

159 """ 

160 handler_response = function(*args, **kwargs) 

161 if self.framework.is_framework_response(handler_response): 

162 return handler_response 

163 elif isinstance(handler_response, ConnexionResponse): 

164 return self.framework.connexion_to_framework_response(handler_response) 

165 else: 

166 return self.build_framework_response(handler_response) 

167 

168 return wrapper 

169 

170 

171class AsyncResponseDecorator(BaseResponseDecorator): 

172 def __call__(self, function: t.Callable) -> t.Callable: 

173 @functools.wraps(function) 

174 async def wrapper(*args, **kwargs): 

175 """ 

176 This method converts a handler response to a framework response. 

177 The handler response can be a ConnexionResponse, a framework response, a tuple or an 

178 object. 

179 """ 

180 handler_response = await function(*args, **kwargs) 

181 if self.framework.is_framework_response(handler_response): 

182 return handler_response 

183 elif isinstance(handler_response, ConnexionResponse): 

184 return self.framework.connexion_to_framework_response(handler_response) 

185 else: 

186 return self.build_framework_response(handler_response) 

187 

188 return wrapper