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