Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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
8import typing as t
10from connexion.utils import all_json
12logger = logging.getLogger("connexion.operations.abstract")
14DEFAULT_MIMETYPE = "application/json"
17class AbstractOperation(metaclass=abc.ABCMeta):
19 """
20 An API routes requests to an Operation by a (path, method) pair.
21 The operation uses a resolver to resolve its handler function.
22 We use the provided spec to do a bunch of heavy lifting before
23 (and after) we call security_schemes handler.
24 The registered handler function ends up looking something like::
26 @secure_endpoint
27 @validate_inputs
28 @deserialize_function_inputs
29 @serialize_function_outputs
30 @validate_outputs
31 def user_provided_handler_function(important, stuff):
32 if important:
33 serious_business(stuff)
34 """
36 def __init__(
37 self,
38 method,
39 path,
40 operation,
41 resolver,
42 app_security=None,
43 security_schemes=None,
44 randomize_endpoint=None,
45 uri_parser_class=None,
46 ):
47 """
48 :param method: HTTP method
49 :type method: str
50 :param path:
51 :type path: str
52 :param operation: swagger operation object
53 :type operation: dict
54 :param resolver: Callable that maps operationID to a function
55 :param app_security: list of security rules the application uses by default
56 :type app_security: list
57 :param security_schemes: `Security Definitions Object
58 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#security-definitions-object>`_
59 :type security_schemes: dict
60 :param randomize_endpoint: number of random characters to append to operation name
61 :type randomize_endpoint: integer
62 :param uri_parser_class: class to use for uri parsing
63 :type uri_parser_class: AbstractURIParser
64 """
65 self._method = method
66 self._path = path
67 self._operation = operation
68 self._resolver = resolver
69 self._security = operation.get("security", app_security)
70 self._security_schemes = security_schemes
71 self._uri_parser_class = uri_parser_class
72 self._randomize_endpoint = randomize_endpoint
73 self._operation_id = self._operation.get("operationId")
75 self._resolution = resolver.resolve(self)
76 self._operation_id = self._resolution.operation_id
78 self._responses = self._operation.get("responses", {})
80 @classmethod
81 @abc.abstractmethod
82 def from_spec(cls, spec, *args, path, method, resolver, **kwargs):
83 pass
85 @property
86 def method(self):
87 """
88 The HTTP method for this operation (ex. GET, POST)
89 """
90 return self._method
92 @property
93 def request_body(self):
94 """The request body for this operation"""
96 @property
97 def is_request_body_defined(self) -> bool:
98 """Whether the request body is defined for this operation"""
99 return self.request_body != {}
101 @property
102 def path(self):
103 """
104 The path of the operation, relative to the API base path
105 """
106 return self._path
108 @property
109 def security(self):
110 return self._security
112 @property
113 def security_schemes(self):
114 return self._security_schemes
116 @property
117 def responses(self):
118 """
119 Returns the responses for this operation
120 """
121 return self._responses
123 @property
124 def operation_id(self):
125 """
126 The operation id used to identify the operation internally to the app
127 """
128 return self._operation_id
130 @property
131 def randomize_endpoint(self):
132 """
133 number of random digits to generate and append to the operation_id.
134 """
135 return self._randomize_endpoint
137 @property
138 def router_controller(self):
139 """
140 The router controller to use (python module where handler functions live)
141 """
142 return self._router_controller
144 @property
145 @abc.abstractmethod
146 def parameters(self):
147 """
148 Returns the parameters for this operation
149 """
151 @property
152 @abc.abstractmethod
153 def produces(self):
154 """
155 Content-Types that the operation produces
156 """
158 @property
159 @abc.abstractmethod
160 def consumes(self):
161 """
162 Content-Types that the operation consumes
163 """
165 @abc.abstractmethod
166 def body_name(self, content_type: str) -> str:
167 """
168 Name of the body in the spec.
169 """
171 @abc.abstractmethod
172 def body_schema(self, content_type: t.Optional[str] = None) -> dict:
173 """
174 The body schema definition for this operation.
175 """
177 @abc.abstractmethod
178 def body_definition(self, content_type: t.Optional[str] = None) -> dict:
179 """
180 The body definition for this operation.
181 :rtype: dict
182 """
184 def response_definition(self, status_code=None, content_type=None):
185 """
186 response definition for this endpoint
187 """
188 response_definition = self.responses.get(
189 str(status_code), self.responses.get("default", {})
190 )
191 return response_definition
193 @abc.abstractmethod
194 def response_schema(self, status_code=None, content_type=None):
195 """
196 response schema for this endpoint
197 """
199 @abc.abstractmethod
200 def example_response(self, status_code=None, content_type=None):
201 """
202 Returns an example from the spec
203 """
205 @abc.abstractmethod
206 def get_path_parameter_types(self):
207 """
208 Returns the types for parameters in the path
209 """
211 @abc.abstractmethod
212 def with_definitions(self, schema):
213 """
214 Returns the given schema, but with the definitions from the spec
215 attached. This allows any remaining references to be resolved by a
216 validator (for example).
217 """
219 def get_mimetype(self):
220 """
221 If the endpoint has no 'produces' then the default is
222 'application/json'.
224 :rtype str
225 """
226 # TODO: don't default
227 if all_json(self.produces):
228 try:
229 return self.produces[0]
230 except IndexError:
231 return DEFAULT_MIMETYPE
232 elif len(self.produces) == 1:
233 return self.produces[0]
234 else:
235 return DEFAULT_MIMETYPE
237 @property
238 def uri_parser_class(self):
239 """
240 The uri parser class for this operation.
241 """
242 return self._uri_parser_class
244 @property
245 def function(self):
246 """
247 Resolved function.
249 :rtype: types.FunctionType
250 """
251 return self._resolution.function