Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/operations/abstract.py: 63%
86 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1"""
2This module defines an AbstractOperation class which implements an abstract Operation interface
3and functionality shared between Swagger 2 and OpenAPI 3 specifications.
4"""
6import abc
7import logging
9from connexion.utils import all_json
11logger = logging.getLogger("connexion.operations.abstract")
13DEFAULT_MIMETYPE = "application/json"
16class AbstractOperation(metaclass=abc.ABCMeta):
18 """
19 An API routes requests to an Operation by a (path, method) pair.
20 The operation uses a resolver to resolve its handler function.
21 We use the provided spec to do a bunch of heavy lifting before
22 (and after) we call security_schemes handler.
23 The registered handler function ends up looking something like::
25 @secure_endpoint
26 @validate_inputs
27 @deserialize_function_inputs
28 @serialize_function_outputs
29 @validate_outputs
30 def user_provided_handler_function(important, stuff):
31 if important:
32 serious_business(stuff)
33 """
35 def __init__(
36 self,
37 method,
38 path,
39 operation,
40 resolver,
41 app_security=None,
42 security_schemes=None,
43 randomize_endpoint=None,
44 uri_parser_class=None,
45 ):
46 """
47 :param method: HTTP method
48 :type method: str
49 :param path:
50 :type path: str
51 :param operation: swagger operation object
52 :type operation: dict
53 :param resolver: Callable that maps operationID to a function
54 :param app_security: list of security rules the application uses by default
55 :type app_security: list
56 :param security_schemes: `Security Definitions Object
57 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#security-definitions-object>`_
58 :type security_schemes: dict
59 :param randomize_endpoint: number of random characters to append to operation name
60 :type randomize_endpoint: integer
61 :param uri_parser_class: class to use for uri parsing
62 :type uri_parser_class: AbstractURIParser
63 """
64 self._method = method
65 self._path = path
66 self._operation = operation
67 self._resolver = resolver
68 self._security = operation.get("security", app_security)
69 self._security_schemes = security_schemes
70 self._uri_parser_class = uri_parser_class
71 self._randomize_endpoint = randomize_endpoint
72 self._operation_id = self._operation.get("operationId")
74 self._resolution = resolver.resolve(self)
75 self._operation_id = self._resolution.operation_id
77 self._responses = self._operation.get("responses", {})
79 @property
80 def method(self):
81 """
82 The HTTP method for this operation (ex. GET, POST)
83 """
84 return self._method
86 @property
87 def request_body(self):
88 """The request body for this operation"""
90 @property
91 def path(self):
92 """
93 The path of the operation, relative to the API base path
94 """
95 return self._path
97 @property
98 def security(self):
99 return self._security
101 @property
102 def security_schemes(self):
103 return self._security_schemes
105 @property
106 def responses(self):
107 """
108 Returns the responses for this operation
109 """
110 return self._responses
112 @property
113 def operation_id(self):
114 """
115 The operation id used to identify the operation internally to the app
116 """
117 return self._operation_id
119 @property
120 def randomize_endpoint(self):
121 """
122 number of random digits to generate and append to the operation_id.
123 """
124 return self._randomize_endpoint
126 @property
127 def router_controller(self):
128 """
129 The router controller to use (python module where handler functions live)
130 """
131 return self._router_controller
133 @property
134 @abc.abstractmethod
135 def parameters(self):
136 """
137 Returns the parameters for this operation
138 """
140 @property
141 @abc.abstractmethod
142 def produces(self):
143 """
144 Content-Types that the operation produces
145 """
147 @property
148 @abc.abstractmethod
149 def consumes(self):
150 """
151 Content-Types that the operation consumes
152 """
154 @abc.abstractmethod
155 def body_name(self, content_type: str) -> str:
156 """
157 Name of the body in the spec.
158 """
160 @abc.abstractmethod
161 def body_schema(self, content_type: str = None) -> dict:
162 """
163 The body schema definition for this operation.
164 """
166 @abc.abstractmethod
167 def body_definition(self, content_type: str = None) -> dict:
168 """
169 The body definition for this operation.
170 :rtype: dict
171 """
173 def response_definition(self, status_code=None, content_type=None):
174 """
175 response definition for this endpoint
176 """
177 response_definition = self.responses.get(
178 str(status_code), self.responses.get("default", {})
179 )
180 return response_definition
182 @abc.abstractmethod
183 def response_schema(self, status_code=None, content_type=None):
184 """
185 response schema for this endpoint
186 """
188 @abc.abstractmethod
189 def example_response(self, status_code=None, content_type=None):
190 """
191 Returns an example from the spec
192 """
194 @abc.abstractmethod
195 def get_path_parameter_types(self):
196 """
197 Returns the types for parameters in the path
198 """
200 @abc.abstractmethod
201 def with_definitions(self, schema):
202 """
203 Returns the given schema, but with the definitions from the spec
204 attached. This allows any remaining references to be resolved by a
205 validator (for example).
206 """
208 def get_mimetype(self):
209 """
210 If the endpoint has no 'produces' then the default is
211 'application/json'.
213 :rtype str
214 """
215 # TODO: don't default
216 if all_json(self.produces):
217 try:
218 return self.produces[0]
219 except IndexError:
220 return DEFAULT_MIMETYPE
221 elif len(self.produces) == 1:
222 return self.produces[0]
223 else:
224 return DEFAULT_MIMETYPE
226 @property
227 def uri_parser_class(self):
228 """
229 The uri parser class for this operation.
230 """
231 return self._uri_parser_class
233 @property
234 def function(self):
235 """
236 Resolved function.
238 :rtype: types.FunctionType
239 """
240 return self._resolution.function