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

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 raise NotImplementedError 

26 

27 @property 

28 def mimetype(self) -> str: 

29 raise NotImplementedError 

30 

31 @property 

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

33 raise NotImplementedError 

34 

35 @property 

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

37 raise NotImplementedError 

38 

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

40 raise NotImplementedError 

41 

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

43 raise NotImplementedError 

44 

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

46 raise NotImplementedError 

47 

48 

49class WSGIRequest(_RequestInterface): 

50 def __init__( 

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

52 ): 

53 self._werkzeug_request = werkzeug_request 

54 self.uri_parser = uri_parser 

55 self.view_args = view_args 

56 

57 self._context = None 

58 self._path_params = None 

59 self._query_params = None 

60 self._form = None 

61 self._body = None 

62 

63 @property 

64 def context(self): 

65 if self._context is None: 

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

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

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

69 return self._context 

70 

71 @property 

72 def content_type(self) -> str: 

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

74 

75 @property 

76 def mimetype(self) -> str: 

77 return self._werkzeug_request.mimetype 

78 

79 @property 

80 def path_params(self): 

81 if self._path_params is None: 

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

83 return self._path_params 

84 

85 @property 

86 def query_params(self): 

87 if self._query_params is None: 

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

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

90 return self._query_params 

91 

92 def form(self): 

93 if self._form is None: 

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

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

96 return self._form 

97 

98 def files(self): 

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

100 

101 def get_body(self): 

102 """Get body based on content type""" 

103 if self._body is None: 

104 if is_json_mimetype(self.content_type): 

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

106 elif self.mimetype in FORM_CONTENT_TYPES: 

107 self._body = self.form() 

108 else: 

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

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

111 return self._body 

112 

113 def __getattr__(self, item): 

114 return getattr(self._werkzeug_request, item) 

115 

116 

117class ASGIRequest(_RequestInterface): 

118 """Wraps starlette Request so it can easily be extended.""" 

119 

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

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

122 self.uri_parser = uri_parser 

123 

124 self._context = None 

125 self._mimetype = None 

126 self._path_params = None 

127 self._query_params = None 

128 self._form = None 

129 self._files = None 

130 

131 @property 

132 def context(self): 

133 if self._context is None: 

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

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

136 return self._context 

137 

138 @property 

139 def content_type(self): 

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

141 

142 @property 

143 def mimetype(self): 

144 if not self._mimetype: 

145 mimetype, _ = parse_options_header(self.content_type) 

146 self._mimetype = mimetype.decode() 

147 return self._mimetype 

148 

149 @property 

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

151 if self._path_params is None: 

152 self._path_params = self.uri_parser.resolve_path( 

153 self._starlette_request.path_params 

154 ) 

155 return self._path_params 

156 

157 @property 

158 def query_params(self): 

159 if self._query_params is None: 

160 args = self._starlette_request.query_params 

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

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

163 return self._query_params 

164 

165 async def form(self): 

166 if self._form is None: 

167 await self._split_form_files() 

168 return self._form 

169 

170 async def files(self): 

171 if self._files is None: 

172 await self._split_form_files() 

173 return self._files 

174 

175 async def _split_form_files(self): 

176 form_data = await self._starlette_request.form() 

177 

178 files = defaultdict(list) 

179 form = defaultdict(list) 

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

181 if isinstance(v, UploadFile): 

182 files[k].append(v) 

183 else: 

184 form[k].append(v) 

185 

186 self._files = files 

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

188 

189 async def json(self): 

190 try: 

191 return await self._starlette_request.json() 

192 except ValueError: 

193 return None 

194 

195 async def get_body(self): 

196 if is_json_mimetype(self.content_type): 

197 return await self.json() 

198 elif self.mimetype in FORM_CONTENT_TYPES: 

199 return await self.form() 

200 else: 

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

202 return await self.body() or None 

203 

204 def __getattr__(self, item): 

205 return getattr(self._starlette_request, item) 

206 

207 

208class ConnexionResponse: 

209 """Connexion interface for a response.""" 

210 

211 def __init__( 

212 self, 

213 status_code=200, 

214 mimetype=None, 

215 content_type=None, 

216 body=None, 

217 headers=None, 

218 is_streamed=False, 

219 ): 

220 self.status_code = status_code 

221 self.mimetype = mimetype 

222 self.content_type = content_type 

223 self.body = body 

224 self.headers = headers or {} 

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

226 self.is_streamed = is_streamed