Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/lifecycle.py: 38%

1""" 

2This module defines interfaces for requests and responses used in Connexion for authentication, 

3validation, serialization, etc. 

4""" 

5import typing as t 

6from collections import defaultdict 

7 

8from multipart.multipart import parse_options_header 

9from starlette.datastructures import UploadFile 

10from starlette.requests import Request as StarletteRequest 

11from werkzeug import Request as WerkzeugRequest 

12 

13from connexion.http_facts import FORM_CONTENT_TYPES 

14from connexion.utils import is_json_mimetype 

15 

16 

17class _RequestInterface: 

18 @property 

19 def context(self) -> t.Dict[str, t.Any]: 

20 """The connexion context of the current request cycle.""" 

21 raise NotImplementedError 

22 

23 @property 

24 def content_type(self) -> str: 

25 """The content type included in the request headers.""" 

26 raise NotImplementedError 

27 

28 @property 

29 def mimetype(self) -> str: 

30 """The content type included in the request headers stripped from any optional character 

31 set encoding""" 

32 raise NotImplementedError 

33 

34 @property 

35 def path_params(self) -> t.Dict[str, t.Any]: 

36 """Path parameters exposed as a dictionary""" 

37 raise NotImplementedError 

38 

39 @property 

40 def query_params(self) -> t.Dict[str, t.Any]: 

41 """Query parameters exposed as a dictionary""" 

42 raise NotImplementedError 

43 

44 def form(self) -> t.Union[t.Dict[str, t.Any], t.Awaitable[t.Dict[str, t.Any]]]: 

45 """Form data, including files.""" 

46 raise NotImplementedError 

47 

48 def files(self) -> t.Dict[str, t.Any]: 

49 """Files included in the request.""" 

50 raise NotImplementedError 

51 

52 def json(self) -> dict: 

53 """Json data included in the request.""" 

54 raise NotImplementedError 

55 

56 def get_body(self) -> t.Any: 

57 """Get body based on the content type. This returns json data for json content types, 

58 form data for form content types, and bytes for all others. If the bytes data is empty, 

59 :code:`None` is returned instead.""" 

60 raise NotImplementedError 

61 

62 

63class WSGIRequest(_RequestInterface): 

64 def __init__( 

65 self, werkzeug_request: WerkzeugRequest, uri_parser=None, view_args=None 

66 ): 

67 self._werkzeug_request = werkzeug_request 

68 self.uri_parser = uri_parser 

69 self.view_args = view_args 

70 

71 self._context = None 

72 self._path_params = None 

73 self._query_params = None 

74 self._form = None 

75 self._body = None 

76 

77 @property 

78 def context(self): 

79 if self._context is None: 

80 scope = self.environ["asgi.scope"] 

81 extensions = scope.setdefault("extensions", {}) 

82 self._context = extensions.setdefault("connexion_context", {}) 

83 return self._context 

84 

85 @property 

86 def content_type(self) -> str: 

87 return self._werkzeug_request.content_type or "application/octet-stream" 

88 

89 @property 

90 def mimetype(self) -> str: 

91 return self._werkzeug_request.mimetype 

92 

93 @property 

94 def path_params(self): 

95 if self._path_params is None: 

96 self._path_params = self.uri_parser.resolve_path(self.view_args) 

97 return self._path_params 

98 

99 @property 

100 def query_params(self): 

101 if self._query_params is None: 

102 query_params = {k: self.args.getlist(k) for k in self.args} 

103 self._query_params = self.uri_parser.resolve_query(query_params) 

104 return self._query_params 

105 

106 def form(self): 

107 if self._form is None: 

108 form = self._werkzeug_request.form.to_dict(flat=False) 

109 self._form = self.uri_parser.resolve_form(form) 

110 return self._form 

111 

112 def files(self): 

113 return self._werkzeug_request.files.to_dict(flat=False) 

114 

115 def json(self): 

116 return self.get_json(silent=True) 

117 

118 def get_body(self): 

119 if self._body is None: 

120 if is_json_mimetype(self.content_type): 

121 self._body = self.get_json(silent=True) 

122 elif self.mimetype in FORM_CONTENT_TYPES: 

123 self._body = self.form() 

124 else: 

125 # Return explicit None instead of empty bytestring so it is handled as null downstream 

126 self._body = self.get_data() or None 

127 return self._body 

128 

129 def __getattr__(self, item): 

130 return getattr(self._werkzeug_request, item) 

131 

132 

133class ConnexionRequest(_RequestInterface): 

134 """ 

135 Implementation of the Connexion :code:`_RequestInterface` representing an ASGI request. 

136 

137 .. attribute:: _starlette_request 

138 :noindex: 

139 

140 This class wraps a Starlette `Request <https://www.starlette.io/requests/#request>`_, 

141 and provides access to its attributes by proxy. 

142 

143 """ 

144 

145 def __init__(self, *args, uri_parser=None, **kwargs): 

146 # Might be set in `from_starlette_request` class method 

147 if not hasattr(self, "_starlette_request"): 

148 self._starlette_request = StarletteRequest(*args, **kwargs) 

149 self.uri_parser = uri_parser 

150 

151 self._context = None 

152 self._mimetype = None 

153 self._path_params = None 

154 self._query_params = None 

155 self._form = None 

156 self._files = None 

157 

158 @classmethod 

159 def from_starlette_request( 

160 cls, request: StarletteRequest, uri_parser=None 

161 ) -> "ConnexionRequest": 

162 # Instantiate the class, and set the `_starlette_request` property before initializing. 

163 self = cls.__new__(cls) 

164 self._starlette_request = request 

165 self.__init__(uri_parser=uri_parser) # type: ignore 

166 return self 

167 

168 @property 

169 def context(self): 

170 if self._context is None: 

171 extensions = self.scope.setdefault("extensions", {}) 

172 self._context = extensions.setdefault("connexion_context", {}) 

173 return self._context 

174 

175 @property 

176 def content_type(self): 

177 return self.headers.get("content-type", "application/octet-stream") 

178 

179 @property 

180 def mimetype(self): 

181 if not self._mimetype: 

182 mimetype, _ = parse_options_header(self.content_type) 

183 self._mimetype = mimetype.decode() 

184 return self._mimetype 

185 

186 @property 

187 def path_params(self) -> t.Dict[str, t.Any]: 

188 if self._path_params is None: 

189 self._path_params = self.uri_parser.resolve_path( 

190 self._starlette_request.path_params 

191 ) 

192 return self._path_params 

193 

194 @property 

195 def query_params(self): 

196 if self._query_params is None: 

197 args = self._starlette_request.query_params 

198 query_params = {k: args.getlist(k) for k in args} 

199 self._query_params = self.uri_parser.resolve_query(query_params) 

200 return self._query_params 

201 

202 async def form(self): 

203 if self._form is None: 

204 await self._split_form_files() 

205 return self._form 

206 

207 async def files(self): 

208 if self._files is None: 

209 await self._split_form_files() 

210 return self._files 

211 

212 async def _split_form_files(self): 

213 form_data = await self._starlette_request.form() 

214 

215 files = defaultdict(list) 

216 form = defaultdict(list) 

217 for k, v in form_data.multi_items(): 

218 if isinstance(v, UploadFile): 

219 files[k].append(v) 

220 else: 

221 form[k].append(v) 

222 

223 self._files = files 

224 self._form = self.uri_parser.resolve_form(form) 

225 

226 async def json(self): 

227 try: 

228 return await self._starlette_request.json() 

229 except ValueError: 

230 return None 

231 

232 async def get_body(self): 

233 if is_json_mimetype(self.content_type): 

234 return await self.json() 

235 elif self.mimetype in FORM_CONTENT_TYPES: 

236 return await self.form() 

237 else: 

238 # Return explicit None instead of empty bytestring so it is handled as null downstream 

239 return await self.body() or None 

240 

241 def __getattr__(self, item): 

242 if self.__getattribute__("_starlette_request"): 

243 return getattr(self._starlette_request, item) 

244 

245 

246class ConnexionResponse: 

247 """Connexion interface for a response.""" 

248 

249 def __init__( 

250 self, 

251 status_code=200, 

252 mimetype=None, 

253 content_type=None, 

254 body=None, 

255 headers=None, 

256 is_streamed=False, 

257 ): 

258 self.status_code = status_code 

259 self.mimetype = mimetype 

260 self.content_type = content_type 

261 self.body = body 

262 self.headers = headers or {} 

263 if content_type: 

264 self.headers.update({"Content-Type": content_type}) 

265 self.is_streamed = is_streamed