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

1""" 

2This module contains resolvers, functions that resolves the user defined view functions 

3from the operations defined in the OpenAPI spec. 

4""" 

5 

6import inspect 

7import logging 

8import typing as t 

9 

10from inflection import camelize 

11 

12import connexion.utils as utils 

13from connexion.exceptions import ResolverError 

14 

15logger = logging.getLogger("connexion.resolver") 

16 

17 

18class Resolution: 

19 def __init__(self, function, operation_id): 

20 """ 

21 Represents the result of operation resolution 

22 

23 :param function: The endpoint function 

24 :type function: types.FunctionType 

25 """ 

26 self.function = function 

27 self.operation_id = operation_id 

28 

29 

30class Resolver: 

31 def __init__(self, function_resolver: t.Callable = utils.get_function_from_name): 

32 """ 

33 Standard resolver 

34 

35 :param function_resolver: Function that resolves functions using an operationId 

36 """ 

37 self.function_resolver = function_resolver 

38 

39 def resolve(self, operation): 

40 """ 

41 Default operation resolver 

42 

43 :type operation: connexion.operations.AbstractOperation 

44 """ 

45 operation_id = self.resolve_operation_id(operation) 

46 return Resolution( 

47 self.resolve_function_from_operation_id(operation_id), operation_id 

48 ) 

49 

50 def resolve_operation_id(self, operation): 

51 """ 

52 Default operationId resolver 

53 

54 :type operation: connexion.operations.AbstractOperation 

55 """ 

56 operation_id = operation.operation_id 

57 router_controller = operation.router_controller 

58 if router_controller is None: 

59 return operation_id 

60 return f"{router_controller}.{operation_id}" 

61 

62 def resolve_function_from_operation_id(self, operation_id): 

63 """ 

64 Invokes the function_resolver 

65 

66 :type operation_id: str 

67 """ 

68 try: 

69 return self.function_resolver(operation_id) 

70 except ImportError as e: 

71 msg = f'Cannot resolve operationId "{operation_id}"! Import error was "{str(e)}"' 

72 raise ResolverError(msg) 

73 except (AttributeError, ValueError) as e: 

74 raise ResolverError(str(e)) 

75 

76 

77class RelativeResolver(Resolver): 

78 """ 

79 Resolves endpoint functions relative to a given root path or module. 

80 """ 

81 

82 def __init__(self, root_path, function_resolver=utils.get_function_from_name): 

83 """ 

84 :param root_path: The root path relative to which an operationId is resolved. 

85 Can also be a module. Has the same effect as setting 

86 `x-swagger-router-controller` or `x-openapi-router-controller` equal to 

87 `root_path` for every operation individually. 

88 :type root_path: typing.Union[str, types.ModuleType] 

89 :param function_resolver: Function that resolves functions using an operationId 

90 :type function_resolver: types.FunctionType 

91 """ 

92 super().__init__(function_resolver=function_resolver) 

93 if inspect.ismodule(root_path): 

94 self.root_path = root_path.__name__ 

95 else: 

96 self.root_path = root_path 

97 

98 def resolve_operation_id(self, operation): 

99 """Resolves the operationId relative to the root path, unless 

100 x-swagger-router-controller or x-openapi-router-controller is specified. 

101 

102 :param operation: The operation to resolve 

103 :type operation: connexion.operations.AbstractOperation 

104 """ 

105 operation_id = operation.operation_id 

106 router_controller = operation.router_controller 

107 if router_controller is None: 

108 return f"{self.root_path}.{operation_id}" 

109 return f"{router_controller}.{operation_id}" 

110 

111 

112class RestyResolver(Resolver): 

113 """ 

114 Resolves endpoint functions using REST semantics (unless overridden by specifying operationId) 

115 """ 

116 

117 def __init__( 

118 self, default_module_name: str, *, collection_endpoint_name: str = "search" 

119 ): 

120 """ 

121 :param default_module_name: Default module name for operations 

122 :param collection_endpoint_name: Name of function to resolve collection endpoints to 

123 """ 

124 super().__init__() 

125 self.default_module_name = default_module_name 

126 self.collection_endpoint_name = collection_endpoint_name 

127 

128 def resolve_operation_id(self, operation): 

129 """ 

130 Resolves the operationId using REST semantics unless explicitly configured in the spec 

131 

132 :type operation: connexion.operations.AbstractOperation 

133 """ 

134 if operation.operation_id: 

135 return super().resolve_operation_id(operation) 

136 

137 return self.resolve_operation_id_using_rest_semantics(operation) 

138 

139 def resolve_operation_id_using_rest_semantics(self, operation): 

140 """ 

141 Resolves the operationId using REST semantics 

142 

143 :type operation: connexion.operations.AbstractOperation 

144 """ 

145 

146 # Split the path into components delimited by '/' 

147 path_components = [c for c in operation.path.split("/") if len(c)] 

148 

149 def is_var(component): 

150 """True if the path component is a var. eg, '{id}'""" 

151 return (component[0] == "{") and (component[-1] == "}") 

152 

153 resource_name = ".".join([c for c in path_components if not is_var(c)]).replace( 

154 "-", "_" 

155 ) 

156 

157 def get_controller_name(): 

158 x_router_controller = operation.router_controller 

159 

160 name = self.default_module_name 

161 

162 if x_router_controller: 

163 name = x_router_controller 

164 

165 elif resource_name: 

166 name += "." + resource_name 

167 

168 return name 

169 

170 def get_function_name(): 

171 method = operation.method 

172 

173 is_collection_endpoint = ( 

174 method.lower() == "get" 

175 and len(resource_name) 

176 and not is_var(path_components[-1]) 

177 ) 

178 

179 return ( 

180 self.collection_endpoint_name 

181 if is_collection_endpoint 

182 else method.lower() 

183 ) 

184 

185 return f"{get_controller_name()}.{get_function_name()}" 

186 

187 

188class MethodResolverBase(RestyResolver): 

189 """ 

190 Resolves endpoint functions based on Flask's MethodView semantics, e.g. 

191 

192 .. code-block:: yaml 

193 

194 paths: 

195 /foo_bar: 

196 get: 

197 # Implied function call: api.FooBarView().get 

198 

199 .. code-block:: python 

200 

201 class FooBarView(MethodView): 

202 def get(self): 

203 return ... 

204 def post(self): 

205 return ... 

206 

207 """ 

208 

209 _class_arguments_type = t.Dict[ 

210 str, t.Dict[str, t.Union[t.Iterable, t.Dict[str, t.Any]]] 

211 ] 

212 

213 def __init__(self, *args, class_arguments: _class_arguments_type = None, **kwargs): 

214 """ 

215 :param args: Arguments passed to :class:`~RestyResolver` 

216 :param class_arguments: Arguments to instantiate the View Class in the format below 

217 :param kwargs: Keywords arguments passed to :class:`~RestyResolver` 

218 

219 .. code-block:: python 

220 

221 { 

222 "ViewName": { 

223 "args": (positional arguments,) 

224 "kwargs": { 

225 "keyword": "argument" 

226 } 

227 } 

228 } 

229 """ 

230 self.class_arguments = class_arguments or {} 

231 super(MethodResolverBase, self).__init__(*args, **kwargs) 

232 self.initialized_views: list = [] 

233 

234 def resolve_operation_id(self, operation): 

235 """ 

236 Resolves the operationId using REST semantics unless explicitly configured in the spec 

237 Once resolved with REST semantics the view_name is capitalised and has 'View' added 

238 to it so it now matches the Class names of the MethodView 

239 

240 :type operation: connexion.operations.AbstractOperation 

241 """ 

242 if operation.operation_id: 

243 # If operation_id is defined then use the higher level API to resolve 

244 return RestyResolver.resolve_operation_id(self, operation) 

245 

246 # Use RestyResolver to get operation_id for us (follow their naming conventions/structure) 

247 operation_id = self.resolve_operation_id_using_rest_semantics(operation) 

248 module_name, view_base, meth_name = operation_id.rsplit(".", 2) 

249 view_name = camelize(view_base) + "View" 

250 

251 return f"{module_name}.{view_name}.{meth_name}" 

252 

253 def resolve_function_from_operation_id(self, operation_id): 

254 """ 

255 Invokes the function_resolver 

256 

257 :type operation_id: str 

258 """ 

259 

260 try: 

261 module_name, view_name, meth_name = operation_id.rsplit(".", 2) 

262 if operation_id and not view_name.endswith("View"): 

263 # If operation_id is not a view then assume it is a standard function 

264 return self.function_resolver(operation_id) 

265 

266 mod = __import__(module_name, fromlist=[view_name]) 

267 view_cls = getattr(mod, view_name) 

268 # find the view and return it 

269 return self.resolve_method_from_class(view_name, meth_name, view_cls) 

270 

271 except ImportError as e: 

272 msg = 'Cannot resolve operationId "{}"! Import error was "{}"'.format( 

273 operation_id, str(e) 

274 ) 

275 raise ResolverError(msg) 

276 except (AttributeError, ValueError) as e: 

277 raise ResolverError(str(e)) 

278 

279 def resolve_method_from_class(self, view_name, meth_name, view_cls): 

280 """ 

281 Returns the view function for the given view class. 

282 """ 

283 raise NotImplementedError() 

284 

285 

286class MethodResolver(MethodResolverBase): 

287 """ 

288 A generic method resolver that instantiates a class and extracts the method 

289 from it, based on the operation id. 

290 """ 

291 

292 def resolve_method_from_class(self, view_name, meth_name, view_cls): 

293 view = None 

294 for v in self.initialized_views: 

295 if v.__class__ == view_cls: 

296 view = v 

297 break 

298 if view is None: 

299 # get the args and kwargs for this view 

300 cls_arguments = self.class_arguments.get(view_name, {}) 

301 cls_args = cls_arguments.get("args", ()) 

302 cls_kwargs = cls_arguments.get("kwargs", {}) 

303 # instantiate the class with the args and kwargs 

304 view = view_cls(*cls_args, **cls_kwargs) 

305 self.initialized_views.append(view) 

306 # get the method if the class 

307 func = getattr(view, meth_name) 

308 # Return the method function of the class 

309 return func 

310 

311 

312class MethodViewResolver(MethodResolverBase): 

313 """ 

314 A specialized method resolver that works with flask's method views. 

315 It resolves the method by calling as_view on the class. 

316 """ 

317 

318 def __init__(self, *args, **kwargs): 

319 if "collection_endpoint_name" in kwargs: 

320 del kwargs["collection_endpoint_name"] 

321 # Dispatch of request is done by Flask 

322 logger.warning( 

323 "collection_endpoint_name is ignored by the MethodViewResolver. " 

324 "Requests to a collection endpoint will be routed to .get()" 

325 ) 

326 super().__init__(*args, **kwargs) 

327 

328 def resolve_method_from_class(self, view_name, meth_name, view_cls): 

329 view = None 

330 for v in self.initialized_views: 

331 # views returned by <class>.as_view 

332 # have the origin class attached as .view_class 

333 if v.view_class == view_cls: 

334 view = v 

335 break 

336 if view is None: 

337 # get the args and kwargs for this view 

338 cls_arguments = self.class_arguments.get(view_name, {}) 

339 cls_args = cls_arguments.get("args", ()) 

340 cls_kwargs = cls_arguments.get("kwargs", {}) 

341 # call as_view to get a view function 

342 # that is decorated with the classes 

343 # decorator list, if any 

344 view = view_cls.as_view(view_name, *cls_args, **cls_kwargs) 

345 # add the view to the list of initialized views 

346 # in order to call as_view only once 

347 self.initialized_views.append(view) 

348 # return the class as view function 

349 # for each operation so that requests 

350 # are dispatched with <class>.dispatch_request, 

351 # when calling the view function 

352 return view