Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/operations/swagger2.py: 20%
150 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-26 06:12 +0000
1"""
2This module defines a Swagger2Operation class, a Connexion operation specific for Swagger 2 specs.
3"""
5import logging
6import typing as t
8from connexion.exceptions import InvalidSpecification
9from connexion.operations.abstract import AbstractOperation
10from connexion.uri_parsing import Swagger2URIParser
11from connexion.utils import deep_get
13logger = logging.getLogger("connexion.operations.swagger2")
16COLLECTION_FORMAT_MAPPING = {
17 "multi": {"style": "form", "explode": True},
18 "csv": {"style": "form", "explode": False},
19 "ssv": {"style": "spaceDelimited", "explode": False},
20 "pipes": {"style": "pipeDelimited", "explode": False},
21}
24class Swagger2Operation(AbstractOperation):
26 """
27 Exposes a Swagger 2.0 operation under the AbstractOperation interface.
28 The primary purpose of this class is to provide the `function()` method
29 to the API. A Swagger2Operation is plugged into the API with the provided
30 (path, method) pair. It resolves the handler function for this operation
31 with the provided resolver, and wraps the handler function with multiple
32 decorators that provide security, validation, serialization,
33 and deserialization.
34 """
36 def __init__(
37 self,
38 method,
39 path,
40 operation,
41 resolver,
42 app_produces,
43 app_consumes,
44 path_parameters=None,
45 app_security=None,
46 security_schemes=None,
47 definitions=None,
48 randomize_endpoint=None,
49 uri_parser_class=None,
50 ):
51 """
52 :param method: HTTP method
53 :type method: str
54 :param path: relative path to this operation
55 :type path: str
56 :param operation: swagger operation object
57 :type operation: dict
58 :param resolver: Callable that maps operationID to a function
59 :type resolver: resolver.Resolver
60 :param app_produces: list of content types the application can return by default
61 :type app_produces: list
62 :param app_consumes: list of content types the application consumes by default
63 :type app_consumes: list
64 :param path_parameters: Parameters defined in the path level
65 :type path_parameters: list
66 :param app_security: list of security rules the application uses by default
67 :type app_security: list
68 :param security_schemes: `Security Definitions Object
69 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#security-definitions-object>`_
70 :type security_schemes: dict
71 :param definitions: `Definitions Object
72 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#definitionsObject>`_
73 :type definitions: dict
74 :param randomize_endpoint: number of random characters to append to operation name
75 :type randomize_endpoint: integer
76 :param uri_parser_class: class to use for uri parsing
77 :type uri_parser_class: AbstractURIParser
78 """
79 uri_parser_class = uri_parser_class or Swagger2URIParser
81 self._router_controller = operation.get("x-swagger-router-controller")
83 super().__init__(
84 method=method,
85 path=path,
86 operation=operation,
87 resolver=resolver,
88 app_security=app_security,
89 security_schemes=security_schemes,
90 randomize_endpoint=randomize_endpoint,
91 uri_parser_class=uri_parser_class,
92 )
94 self._produces = operation.get("produces", app_produces)
95 self._consumes = operation.get("consumes", app_consumes)
97 self.definitions = definitions or {}
99 self._parameters = operation.get("parameters", [])
100 if path_parameters:
101 self._parameters += path_parameters
103 self._responses = operation.get("responses", {})
105 @classmethod
106 def from_spec(cls, spec, *args, path, method, resolver, **kwargs):
107 return cls(
108 method,
109 path,
110 spec.get_operation(path, method),
111 resolver=resolver,
112 path_parameters=spec.get_path_params(path),
113 app_produces=spec.produces,
114 app_consumes=spec.consumes,
115 app_security=spec.security,
116 security_schemes=spec.security_schemes,
117 definitions=spec.definitions,
118 *args,
119 **kwargs,
120 )
122 @property
123 def request_body(self) -> dict:
124 if not hasattr(self, "_request_body"):
125 body_params = []
126 form_params = []
127 for parameter in self.parameters:
128 if parameter["in"] == "body":
129 body_params.append(parameter)
130 elif parameter["in"] == "formData":
131 form_params.append(parameter)
133 if len(body_params) > 1:
134 raise InvalidSpecification(
135 f"{self.method} {self.path}: There can be one 'body' parameter at most"
136 )
138 if body_params and form_params:
139 raise InvalidSpecification(
140 f"{self.method} {self.path}: 'body' and 'formData' parameters are mutually exclusive"
141 )
143 if body_params:
144 self._request_body = self._transform_json(body_params[0])
145 elif form_params:
146 self._request_body = self._transform_form(form_params)
147 else:
148 self._request_body = {}
150 return self._request_body
152 @property
153 def parameters(self):
154 return self._parameters
156 @property
157 def consumes(self):
158 return self._consumes
160 @property
161 def produces(self):
162 return self._produces
164 def get_path_parameter_types(self):
165 types = {}
166 path_parameters = (p for p in self.parameters if p["in"] == "path")
167 for path_defn in path_parameters:
168 if path_defn.get("type") == "string" and path_defn.get("format") == "path":
169 # path is special case for type 'string'
170 path_type = "path"
171 else:
172 path_type = path_defn.get("type")
173 types[path_defn["name"]] = path_type
174 return types
176 def with_definitions(self, schema):
177 if "schema" in schema:
178 schema["schema"]["definitions"] = self.definitions
179 return schema
181 def response_schema(self, status_code=None, content_type=None):
182 response_definition = self.response_definition(status_code, content_type)
183 return self.with_definitions(response_definition.get("schema", {}))
185 def example_response(self, status_code=None, *args, **kwargs):
186 """
187 Returns example response from spec
188 """
189 # simply use the first/lowest status code, this is probably 200 or 201
190 status_code = status_code or sorted(self._responses.keys())[0]
191 examples_path = [str(status_code), "examples"]
192 schema_example_path = [str(status_code), "schema", "example"]
193 schema_path = [str(status_code), "schema"]
195 try:
196 status_code = int(status_code)
197 except ValueError:
198 status_code = 200
199 try:
200 return (
201 list(deep_get(self._responses, examples_path).values())[0],
202 status_code,
203 )
204 except KeyError:
205 pass
206 try:
207 return (deep_get(self._responses, schema_example_path), status_code)
208 except KeyError:
209 pass
211 try:
212 return (
213 self._nested_example(deep_get(self._responses, schema_path)),
214 status_code,
215 )
216 except KeyError:
217 return (None, status_code)
219 def _nested_example(self, schema):
220 try:
221 return schema["example"]
222 except KeyError:
223 pass
224 try:
225 # Recurse if schema is an object
226 return {
227 key: self._nested_example(value)
228 for (key, value) in schema["properties"].items()
229 }
230 except KeyError:
231 pass
232 try:
233 # Recurse if schema is an array
234 return [self._nested_example(schema["items"])]
235 except KeyError:
236 raise
238 def body_name(self, content_type: str = None) -> str:
239 return self.body_definition(content_type).get("name", "body")
241 def body_schema(self, content_type: str = None) -> dict:
242 """
243 The body schema definition for this operation.
244 """
245 body_definition = self.body_definition(content_type)
246 return self.with_definitions(body_definition).get("schema", {})
248 def body_definition(self, content_type: str = None) -> dict:
249 """
250 The body complete definition for this operation.
252 **There can be one "body" parameter at most.**
253 """
254 return self.request_body
256 def _transform_json(self, body_parameter: dict) -> dict:
257 """Translate Swagger2 json parameters into OpenAPI 3 jsonschema spec."""
258 nullable = body_parameter.get("x-nullable")
259 if nullable is not None:
260 body_parameter["schema"]["nullable"] = nullable
261 return body_parameter
263 def _transform_form(self, form_parameters: t.List[dict]) -> dict:
264 """Translate Swagger2 form parameters into OpenAPI 3 jsonschema spec."""
265 properties = {}
266 defaults = {}
267 required = []
268 encoding = {}
270 for param in form_parameters:
271 prop = {}
273 if param["type"] == "file":
274 prop.update(
275 {
276 "type": "string",
277 "format": "binary",
278 }
279 )
280 else:
281 prop["type"] = param["type"]
283 format_ = param.get("format")
284 if format_ is not None:
285 prop["format"] = format_
287 default = param.get("default")
288 if default is not None:
289 prop["default"] = default
290 defaults[param["name"]] = default
292 nullable = param.get("x-nullable")
293 if nullable is not None:
294 prop["nullable"] = nullable
296 if param["type"] == "array":
297 prop["items"] = param.get("items", {})
299 collection_format = param.get("collectionFormat", "csv")
300 try:
301 encoding[param["name"]] = COLLECTION_FORMAT_MAPPING[
302 collection_format
303 ]
304 except KeyError:
305 raise InvalidSpecification(
306 f"The collection format ({collection_format}) is not supported by "
307 f"Connexion as it cannot be mapped to OpenAPI 3."
308 )
310 properties[param["name"]] = prop
312 if param.get("required", False):
313 required.append(param["name"])
315 definition: t.Dict[str, t.Any] = {
316 "schema": {
317 "type": "object",
318 "properties": properties,
319 "required": required,
320 }
321 }
322 if defaults:
323 definition["schema"]["default"] = defaults
324 if encoding:
325 definition["encoding"] = encoding
327 return definition