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

1""" 

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

3validation, serialization, etc. 

4""" 

5 

6import typing as t 

7from collections import defaultdict 

8 

9from python_multipart.multipart import parse_options_header 

10from starlette.datastructures import UploadFile 

11from starlette.requests import Request as StarletteRequest 

12from werkzeug import Request as WerkzeugRequest 

13 

14from connexion.http_facts import FORM_CONTENT_TYPES 

15from connexion.utils import is_json_mimetype 

16 

17 

18class _RequestInterface: 

19 @property 

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

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

22 raise NotImplementedError 

23 

24 @property 

25 def content_type(self) -> str: 

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

27 raise NotImplementedError 

28 

29 @property 

30 def mimetype(self) -> str: 

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

32 set encoding""" 

33 raise NotImplementedError 

34 

35 @property 

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

37 """Path parameters exposed as a dictionary""" 

38 raise NotImplementedError 

39 

40 @property 

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

42 """Query parameters exposed as a dictionary""" 

43 raise NotImplementedError 

44 

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

46 """Form data, including files.""" 

47 raise NotImplementedError 

48 

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

50 """Files included in the request.""" 

51 raise NotImplementedError 

52 

53 def json(self) -> dict: 

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

55 raise NotImplementedError 

56 

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

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

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

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

61 raise NotImplementedError 

62 

63 

64class WSGIRequest(_RequestInterface): 

65 def __init__( 

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

67 ): 

68 self._werkzeug_request = werkzeug_request 

69 self.uri_parser = uri_parser 

70 self.view_args = view_args 

71 

72 self._context = None 

73 self._path_params = None 

74 self._query_params = None 

75 self._form = None 

76 self._body = None 

77 

78 @property 

79 def context(self): 

80 if self._context is None: 

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

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

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

84 return self._context 

85 

86 @property 

87 def content_type(self) -> str: 

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

89 

90 @property 

91 def mimetype(self) -> str: 

92 return self._werkzeug_request.mimetype 

93 

94 @property 

95 def path_params(self): 

96 if self._path_params is None: 

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

98 return self._path_params 

99 

100 @property 

101 def query_params(self): 

102 if self._query_params is None: 

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

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

105 return self._query_params 

106 

107 def form(self): 

108 if self._form is None: 

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

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

111 return self._form 

112 

113 def files(self): 

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

115 

116 def json(self): 

117 return self.get_json(silent=True) 

118 

119 def get_body(self): 

120 if self._body is None: 

121 if is_json_mimetype(self.content_type): 

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

123 elif self.mimetype in FORM_CONTENT_TYPES: 

124 self._body = self.form() 

125 else: 

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

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

128 return self._body 

129 

130 def __getattr__(self, item): 

131 return getattr(self._werkzeug_request, item) 

132 

133 

134class ConnexionRequest(_RequestInterface): 

135 """ 

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

137 

138 .. attribute:: _starlette_request 

139 :noindex: 

140 

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

142 and provides access to its attributes by proxy. 

143 

144 """ 

145 

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

147 # Might be set in `from_starlette_request` class method 

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

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

150 self.uri_parser = uri_parser 

151 

152 self._context = None 

153 self._mimetype = None 

154 self._path_params = None 

155 self._query_params = None 

156 self._form = None 

157 self._files = None 

158 

159 @classmethod 

160 def from_starlette_request( 

161 cls, request: StarletteRequest, uri_parser=None 

162 ) -> "ConnexionRequest": 

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

164 self = cls.__new__(cls) 

165 self._starlette_request = request 

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

167 return self 

168 

169 @property 

170 def context(self): 

171 if self._context is None: 

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

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

174 return self._context 

175 

176 @property 

177 def content_type(self): 

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

179 

180 @property 

181 def mimetype(self): 

182 if not self._mimetype: 

183 mimetype, _ = parse_options_header(self.content_type) 

184 self._mimetype = mimetype.decode() 

185 return self._mimetype 

186 

187 @property 

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

189 if self._path_params is None: 

190 self._path_params = self.uri_parser.resolve_path( 

191 self._starlette_request.path_params 

192 ) 

193 return self._path_params 

194 

195 @property 

196 def query_params(self): 

197 if self._query_params is None: 

198 args = self._starlette_request.query_params 

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

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

201 return self._query_params 

202 

203 async def form(self): 

204 if self._form is None: 

205 await self._split_form_files() 

206 return self._form 

207 

208 async def files(self): 

209 if self._files is None: 

210 await self._split_form_files() 

211 return self._files 

212 

213 async def _split_form_files(self): 

214 form_data = await self._starlette_request.form() 

215 

216 files = defaultdict(list) 

217 form = defaultdict(list) 

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

219 if isinstance(v, UploadFile): 

220 files[k].append(v) 

221 else: 

222 form[k].append(v) 

223 

224 self._files = files 

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

226 

227 async def json(self): 

228 try: 

229 return await self._starlette_request.json() 

230 except ValueError: 

231 return None 

232 

233 async def get_body(self): 

234 if is_json_mimetype(self.content_type): 

235 return await self.json() 

236 elif self.mimetype in FORM_CONTENT_TYPES: 

237 return await self.form() 

238 else: 

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

240 return await self.body() or None 

241 

242 def __getattr__(self, item): 

243 if self.__getattribute__("_starlette_request"): 

244 return getattr(self._starlette_request, item) 

245 

246 

247class ConnexionResponse: 

248 """Connexion interface for a response.""" 

249 

250 def __init__( 

251 self, 

252 status_code=200, 

253 mimetype=None, 

254 content_type=None, 

255 body=None, 

256 headers=None, 

257 is_streamed=False, 

258 ): 

259 self.status_code = status_code 

260 self.mimetype = mimetype 

261 self.content_type = content_type 

262 self.body = body 

263 self.headers = headers or {} 

264 if content_type: 

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

266 self.is_streamed = is_streamed