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 OpenAPIOperation class, a Connexion operation specific for OpenAPI 3 specs.
3"""
5import logging
7from connexion.datastructures import MediaTypeDict
8from connexion.operations.abstract import AbstractOperation
9from connexion.uri_parsing import OpenAPIURIParser
10from connexion.utils import build_example_from_schema, deep_get
12logger = logging.getLogger("connexion.operations.openapi3")
15class OpenAPIOperation(AbstractOperation):
17 """
18 A single API operation on a path.
19 """
21 def __init__(
22 self,
23 method,
24 path,
25 operation,
26 resolver,
27 path_parameters=None,
28 app_security=None,
29 security_schemes=None,
30 components=None,
31 randomize_endpoint=None,
32 uri_parser_class=None,
33 ):
34 """
35 This class uses the OperationID identify the module and function that will handle the operation
37 From Swagger Specification:
39 **OperationID**
41 A friendly name for the operation. The id MUST be unique among all operations described in the API.
42 Tools and libraries MAY use the operation id to uniquely identify an operation.
44 :param method: HTTP method
45 :type method: str
46 :param path:
47 :type path: str
48 :param operation: swagger operation object
49 :type operation: dict
50 :param resolver: Callable that maps operationID to a function
51 :param path_parameters: Parameters defined in the path level
52 :type path_parameters: list
53 :param app_security: list of security rules the application uses by default
54 :type app_security: list
55 :param security_schemes: `Security Definitions Object
56 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#security-definitions-object>`_
57 :type security_schemes: dict
58 :param components: `Components Object
59 <https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.1.md#componentsObject>`_
60 :type components: dict
61 :param randomize_endpoint: number of random characters to append to operation name
62 :type randomize_endpoint: integer
63 :param uri_parser_class: class to use for uri parsing
64 :type uri_parser_class: AbstractURIParser
65 """
66 self.components = components or {}
68 uri_parser_class = uri_parser_class or OpenAPIURIParser
70 self._router_controller = operation.get("x-openapi-router-controller")
72 super().__init__(
73 method=method,
74 path=path,
75 operation=operation,
76 resolver=resolver,
77 app_security=app_security,
78 security_schemes=security_schemes,
79 randomize_endpoint=randomize_endpoint,
80 uri_parser_class=uri_parser_class,
81 )
83 self._parameters = operation.get("parameters", [])
84 if path_parameters:
85 self._parameters += path_parameters
87 self._responses = operation.get("responses", {})
89 # TODO figure out how to support multiple mimetypes
90 # NOTE we currently just combine all of the possible mimetypes,
91 # but we need to refactor to support mimetypes by response code
92 response_content_types = []
93 for _, defn in self._responses.items():
94 response_content_types += defn.get("content", {}).keys()
95 self._produces = response_content_types
96 self._consumes = None
98 logger.debug("consumes: %s" % self.consumes)
99 logger.debug("produces: %s" % self.produces)
101 @classmethod
102 def from_spec(cls, spec, *args, path, method, resolver, **kwargs):
103 return cls(
104 method,
105 path,
106 spec.get_operation(path, method),
107 resolver=resolver,
108 path_parameters=spec.get_path_params(path),
109 app_security=spec.security,
110 security_schemes=spec.security_schemes,
111 components=spec.components,
112 *args,
113 **kwargs,
114 )
116 @property
117 def request_body(self):
118 return self._operation.get("requestBody", {})
120 @property
121 def parameters(self):
122 return self._parameters
124 @property
125 def consumes(self):
126 if self._consumes is None:
127 request_content = self.request_body.get("content", {})
128 self._consumes = list(request_content.keys())
129 return self._consumes
131 @property
132 def produces(self):
133 return self._produces
135 def with_definitions(self, schema: dict):
136 if self.components:
137 schema.setdefault("schema", {})
138 schema["schema"]["components"] = self.components
139 return schema
141 def response_schema(self, status_code=None, content_type=None):
142 response_definition = self.response_definition(status_code, content_type)
143 content_definition = response_definition.get("content", response_definition)
144 content_definition = content_definition.get(content_type, content_definition)
145 if "schema" in content_definition:
146 return self.with_definitions(content_definition).get("schema", {})
147 return {}
149 def example_response(self, status_code=None, content_type=None):
150 """
151 Returns example response from spec
152 """
153 # simply use the first/lowest status code, this is probably 200 or 201
154 status_code = status_code or sorted(self._responses.keys())[0]
156 content_type = content_type or self.get_mimetype()
157 examples_path = [str(status_code), "content", content_type, "examples"]
158 example_path = [str(status_code), "content", content_type, "example"]
159 schema_example_path = [
160 str(status_code),
161 "content",
162 content_type,
163 "schema",
164 "example",
165 ]
166 schema_path = [str(status_code), "content", content_type, "schema"]
168 try:
169 status_code = int(status_code)
170 except ValueError:
171 status_code = 200
172 try:
173 # TODO also use example header?
174 return (
175 list(deep_get(self._responses, examples_path).values())[0]["value"],
176 status_code,
177 )
178 except (KeyError, IndexError):
179 pass
180 try:
181 return (deep_get(self._responses, example_path), status_code)
182 except KeyError:
183 pass
184 try:
185 return (deep_get(self._responses, schema_example_path), status_code)
186 except KeyError:
187 pass
189 try:
190 schema = deep_get(self._responses, schema_path)
191 except KeyError:
192 return ("No example response or response schema defined.", status_code)
194 return (build_example_from_schema(schema), status_code)
196 def get_path_parameter_types(self):
197 types = {}
198 path_parameters = (p for p in self.parameters if p["in"] == "path")
199 for path_defn in path_parameters:
200 path_schema = path_defn["schema"]
201 if (
202 path_schema.get("type") == "string"
203 and path_schema.get("format") == "path"
204 ):
205 # path is special case for type 'string'
206 path_type = "path"
207 else:
208 path_type = path_schema.get("type")
209 types[path_defn["name"]] = path_type
210 return types
212 def body_name(self, _content_type: str) -> str:
213 return self.request_body.get("x-body-name", "body")
215 def body_schema(self, content_type: str = None) -> dict:
216 """
217 The body schema definition for this operation.
218 """
219 return self.body_definition(content_type).get("schema", {})
221 def body_definition(self, content_type: str = None) -> dict:
222 """
223 The body complete definition for this operation.
225 **There can be one "body" parameter at most.**
226 """
227 if self.request_body:
228 if content_type is None:
229 # TODO: make content type required
230 content_type = self.consumes[0]
231 if len(self.consumes) > 1:
232 logger.warning(
233 "this operation accepts multiple content types, using %s",
234 content_type,
235 )
236 content_type_dict = MediaTypeDict(self.request_body.get("content", {}))
237 res = content_type_dict.get(content_type, {})
238 return self.with_definitions(res)
239 return {}