Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/resolver.py: 24%
130 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1"""
2This module contains resolvers, functions that resolves the user defined view functions
3from the operations defined in the OpenAPI spec.
4"""
6import inspect
7import logging
8import typing as t
10from inflection import camelize
12import connexion.utils as utils
13from connexion.exceptions import ResolverError
15logger = logging.getLogger("connexion.resolver")
18class Resolution:
19 def __init__(self, function, operation_id):
20 """
21 Represents the result of operation resolution
23 :param function: The endpoint function
24 :type function: types.FunctionType
25 """
26 self.function = function
27 self.operation_id = operation_id
30class Resolver:
31 def __init__(self, function_resolver: t.Callable = utils.get_function_from_name):
32 """
33 Standard resolver
35 :param function_resolver: Function that resolves functions using an operationId
36 """
37 self.function_resolver = function_resolver
39 def resolve(self, operation):
40 """
41 Default operation resolver
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 )
50 def resolve_operation_id(self, operation):
51 """
52 Default operationId resolver
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}"
62 def resolve_function_from_operation_id(self, operation_id):
63 """
64 Invokes the function_resolver
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))
77class RelativeResolver(Resolver):
78 """
79 Resolves endpoint functions relative to a given root path or module.
80 """
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
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.
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}"
112class RestyResolver(Resolver):
113 """
114 Resolves endpoint functions using REST semantics (unless overridden by specifying operationId)
115 """
117 def __init__(self, default_module_name, collection_endpoint_name="search"):
118 """
119 :param default_module_name: Default module name for operations
120 :type default_module_name: str
121 """
122 super().__init__()
123 self.default_module_name = default_module_name
124 self.collection_endpoint_name = collection_endpoint_name
126 def resolve_operation_id(self, operation):
127 """
128 Resolves the operationId using REST semantics unless explicitly configured in the spec
130 :type operation: connexion.operations.AbstractOperation
131 """
132 if operation.operation_id:
133 return super().resolve_operation_id(operation)
135 return self.resolve_operation_id_using_rest_semantics(operation)
137 def resolve_operation_id_using_rest_semantics(self, operation):
138 """
139 Resolves the operationId using REST semantics
141 :type operation: connexion.operations.AbstractOperation
142 """
144 # Split the path into components delimited by '/'
145 path_components = [c for c in operation.path.split("/") if len(c)]
147 def is_var(component):
148 """True if the path component is a var. eg, '{id}'"""
149 return (component[0] == "{") and (component[-1] == "}")
151 resource_name = ".".join([c for c in path_components if not is_var(c)]).replace(
152 "-", "_"
153 )
155 def get_controller_name():
156 x_router_controller = operation.router_controller
158 name = self.default_module_name
160 if x_router_controller:
161 name = x_router_controller
163 elif resource_name:
164 name += "." + resource_name
166 return name
168 def get_function_name():
169 method = operation.method
171 is_collection_endpoint = (
172 method.lower() == "get"
173 and len(resource_name)
174 and not is_var(path_components[-1])
175 )
177 return (
178 self.collection_endpoint_name
179 if is_collection_endpoint
180 else method.lower()
181 )
183 return f"{get_controller_name()}.{get_function_name()}"
186class MethodResolverBase(RestyResolver):
187 """
188 Resolves endpoint functions based on Flask's MethodView semantics, e.g. ::
190 paths:
191 /foo_bar:
192 get:
193 # Implied function call: api.FooBarView().get
195 class FooBarView(MethodView):
196 def get(self):
197 return ...
198 def post(self):
199 return ...
200 """
202 _class_arguments_type = t.Dict[
203 str, t.Dict[str, t.Union[t.Iterable, t.Dict[str, t.Any]]]
204 ]
206 def __init__(self, *args, class_arguments: _class_arguments_type = None, **kwargs):
207 """
208 :param class_arguments: Arguments to instantiate the View Class in the format # noqa
209 {
210 "ViewName": {
211 "args": (positional arguments,)
212 "kwargs": {
213 "keyword": "argument"
214 }
215 }
216 }
217 """
218 self.class_arguments = class_arguments or {}
219 if "collection_endpoint_name" in kwargs:
220 del kwargs["collection_endpoint_name"]
221 # Dispatch of request is done by Flask
222 logger.warning(
223 "collection_endpoint_name is ignored by the MethodViewResolver. "
224 "Requests to a collection endpoint will be routed to .get()"
225 )
226 super(MethodResolverBase, self).__init__(*args, **kwargs)
227 self.initialized_views: list = []
229 def resolve_operation_id(self, operation):
230 """
231 Resolves the operationId using REST semantics unless explicitly configured in the spec
232 Once resolved with REST semantics the view_name is capitalised and has 'View' added
233 to it so it now matches the Class names of the MethodView
235 :type operation: connexion.operations.AbstractOperation
236 """
237 if operation.operation_id:
238 # If operation_id is defined then use the higher level API to resolve
239 return RestyResolver.resolve_operation_id(self, operation)
241 # Use RestyResolver to get operation_id for us (follow their naming conventions/structure)
242 operation_id = self.resolve_operation_id_using_rest_semantics(operation)
243 module_name, view_base, meth_name = operation_id.rsplit(".", 2)
244 view_name = camelize(view_base) + "View"
246 return f"{module_name}.{view_name}.{meth_name}"
248 def resolve_function_from_operation_id(self, operation_id):
249 """
250 Invokes the function_resolver
252 :type operation_id: str
253 """
255 try:
256 module_name, view_name, meth_name = operation_id.rsplit(".", 2)
257 if operation_id and not view_name.endswith("View"):
258 # If operation_id is not a view then assume it is a standard function
259 return self.function_resolver(operation_id)
261 mod = __import__(module_name, fromlist=[view_name])
262 view_cls = getattr(mod, view_name)
263 # find the view and return it
264 return self.resolve_method_from_class(view_name, meth_name, view_cls)
266 except ImportError as e:
267 msg = 'Cannot resolve operationId "{}"! Import error was "{}"'.format(
268 operation_id, str(e)
269 )
270 raise ResolverError(msg)
271 except (AttributeError, ValueError) as e:
272 raise ResolverError(str(e))
274 def resolve_method_from_class(self, view_name, meth_name, view_cls):
275 """
276 Returns the view function for the given view class.
277 """
278 raise NotImplementedError()
281class MethodResolver(MethodResolverBase):
282 """
283 A generic method resolver that instantiates a class a extracts the method
284 from it, based on the operation id.
285 """
287 def resolve_method_from_class(self, view_name, meth_name, view_cls):
288 view = None
289 for v in self.initialized_views:
290 if v.__class__ == view_cls:
291 view = v
292 break
293 if view is None:
294 # get the args and kwargs for this view
295 cls_arguments = self.class_arguments.get(view_name, {})
296 cls_args = cls_arguments.get("args", ())
297 cls_kwargs = cls_arguments.get("kwargs", {})
298 # instantiate the class with the args and kwargs
299 view = view_cls(*cls_args, **cls_kwargs)
300 self.initialized_views.append(view)
301 # get the method if the class
302 func = getattr(view, meth_name)
303 # Return the method function of the class
304 return func
307class MethodViewResolver(MethodResolverBase):
308 """
309 A specialized method resolver that works with flask's method views.
310 It resolves the method by calling as_view on the class.
311 """
313 def resolve_method_from_class(self, view_name, meth_name, view_cls):
314 view = None
315 for v in self.initialized_views:
316 # views returned by <class>.as_view
317 # have the origin class attached as .view_class
318 if v.view_class == view_cls:
319 view = v
320 break
321 if view is None:
322 # get the args and kwargs for this view
323 cls_arguments = self.class_arguments.get(view_name, {})
324 cls_args = cls_arguments.get("args", ())
325 cls_kwargs = cls_arguments.get("kwargs", {})
326 # call as_view to get a view function
327 # that is decorated with the classes
328 # decorator list, if any
329 view = view_cls.as_view(view_name, *cls_args, **cls_kwargs)
330 # add the view to the list of initialized views
331 # in order to call as_view only once
332 self.initialized_views.append(view)
333 # return the class as view function
334 # for each operation so that requests
335 # are dispatched with <class>.dispatch_request,
336 # when calling the view function
337 return view