Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/operations/openapi.py: 22%
117 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
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 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 return (
191 self._nested_example(deep_get(self._responses, schema_path)),
192 status_code,
193 )
194 except KeyError:
195 return (None, status_code)
197 def _nested_example(self, schema):
198 try:
199 return schema["example"]
200 except KeyError:
201 pass
202 try:
203 # Recurse if schema is an object
204 return {
205 key: self._nested_example(value)
206 for (key, value) in schema["properties"].items()
207 }
208 except KeyError:
209 pass
210 try:
211 # Recurse if schema is an array
212 return [self._nested_example(schema["items"])]
213 except KeyError:
214 raise
216 def get_path_parameter_types(self):
217 types = {}
218 path_parameters = (p for p in self.parameters if p["in"] == "path")
219 for path_defn in path_parameters:
220 path_schema = path_defn["schema"]
221 if (
222 path_schema.get("type") == "string"
223 and path_schema.get("format") == "path"
224 ):
225 # path is special case for type 'string'
226 path_type = "path"
227 else:
228 path_type = path_schema.get("type")
229 types[path_defn["name"]] = path_type
230 return types
232 def body_name(self, _content_type: str) -> str:
233 return self.request_body.get("x-body-name", "body")
235 def body_schema(self, content_type: str = None) -> dict:
236 """
237 The body schema definition for this operation.
238 """
239 return self.body_definition(content_type).get("schema", {})
241 def body_definition(self, content_type: str = None) -> dict:
242 """
243 The body complete definition for this operation.
245 **There can be one "body" parameter at most.**
246 """
247 if self.request_body:
248 if content_type is None:
249 # TODO: make content type required
250 content_type = self.consumes[0]
251 if len(self.consumes) > 1:
252 logger.warning(
253 "this operation accepts multiple content types, using %s",
254 content_type,
255 )
256 content_type_dict = MediaTypeDict(self.request_body.get("content", {}))
257 res = content_type_dict.get(content_type, {})
258 return self.with_definitions(res)
259 return {}