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

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__( 

214 self, *args, class_arguments: t.Optional[_class_arguments_type] = None, **kwargs 

215 ): 

216 """ 

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

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

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

220 

221 .. code-block:: python 

222 

223 { 

224 "ViewName": { 

225 "args": (positional arguments,) 

226 "kwargs": { 

227 "keyword": "argument" 

228 } 

229 } 

230 } 

231 """ 

232 self.class_arguments = class_arguments or {} 

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

234 self.initialized_views: list = [] 

235 

236 def resolve_operation_id(self, operation): 

237 """ 

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

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

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

241 

242 :type operation: connexion.operations.AbstractOperation 

243 """ 

244 if operation.operation_id: 

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

246 return RestyResolver.resolve_operation_id(self, operation) 

247 

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

249 operation_id = self.resolve_operation_id_using_rest_semantics(operation) 

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

251 view_name = camelize(view_base) + "View" 

252 

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

254 

255 def resolve_function_from_operation_id(self, operation_id): 

256 """ 

257 Invokes the function_resolver 

258 

259 :type operation_id: str 

260 """ 

261 

262 try: 

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

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

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

266 return self.function_resolver(operation_id) 

267 

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

269 view_cls = getattr(mod, view_name) 

270 # find the view and return it 

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

272 

273 except ImportError as e: 

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

275 operation_id, str(e) 

276 ) 

277 raise ResolverError(msg) 

278 except (AttributeError, ValueError) as e: 

279 raise ResolverError(str(e)) 

280 

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

282 """ 

283 Returns the view function for the given view class. 

284 """ 

285 raise NotImplementedError() 

286 

287 

288class MethodResolver(MethodResolverBase): 

289 """ 

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

291 from it, based on the operation id. 

292 """ 

293 

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

295 view = None 

296 for v in self.initialized_views: 

297 if v.__class__ == view_cls: 

298 view = v 

299 break 

300 if view is None: 

301 # get the args and kwargs for this view 

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

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

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

305 # instantiate the class with the args and kwargs 

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

307 self.initialized_views.append(view) 

308 # get the method if the class 

309 func = getattr(view, meth_name) 

310 # Return the method function of the class 

311 return func 

312 

313 

314class MethodViewResolver(MethodResolverBase): 

315 """ 

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

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

318 """ 

319 

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

321 if "collection_endpoint_name" in kwargs: 

322 del kwargs["collection_endpoint_name"] 

323 # Dispatch of request is done by Flask 

324 logger.warning( 

325 "collection_endpoint_name is ignored by the MethodViewResolver. " 

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

327 ) 

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

329 

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

331 view = None 

332 for v in self.initialized_views: 

333 # views returned by <class>.as_view 

334 # have the origin class attached as .view_class 

335 if v.view_class == view_cls: 

336 view = v 

337 break 

338 if view is None: 

339 # get the args and kwargs for this view 

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

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

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

343 # call as_view to get a view function 

344 # that is decorated with the classes 

345 # decorator list, if any 

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

347 # add the view to the list of initialized views 

348 # in order to call as_view only once 

349 self.initialized_views.append(view) 

350 # return the class as view function 

351 # for each operation so that requests 

352 # are dispatched with <class>.dispatch_request, 

353 # when calling the view function 

354 return view