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 a Swagger2Operation class, a Connexion operation specific for Swagger 2 specs.
3"""
5import logging
6import typing as t
7from http import HTTPStatus
9from connexion.datastructures import NoContent
10from connexion.exceptions import InvalidSpecification
11from connexion.operations.abstract import AbstractOperation
12from connexion.uri_parsing import Swagger2URIParser
13from connexion.utils import build_example_from_schema, deep_get
15logger = logging.getLogger("connexion.operations.swagger2")
18COLLECTION_FORMAT_MAPPING = {
19 "multi": {"style": "form", "explode": True},
20 "csv": {"style": "form", "explode": False},
21 "ssv": {"style": "spaceDelimited", "explode": False},
22 "pipes": {"style": "pipeDelimited", "explode": False},
23}
26class Swagger2Operation(AbstractOperation):
27 """
28 Exposes a Swagger 2.0 operation under the AbstractOperation interface.
29 The primary purpose of this class is to provide the `function()` method
30 to the API. A Swagger2Operation is plugged into the API with the provided
31 (path, method) pair. It resolves the handler function for this operation
32 with the provided resolver, and wraps the handler function with multiple
33 decorators that provide security, validation, serialization,
34 and deserialization.
35 """
37 def __init__(
38 self,
39 method,
40 path,
41 operation,
42 resolver,
43 app_produces,
44 app_consumes,
45 path_parameters=None,
46 app_security=None,
47 security_schemes=None,
48 definitions=None,
49 randomize_endpoint=None,
50 uri_parser_class=None,
51 ):
52 """
53 :param method: HTTP method
54 :type method: str
55 :param path: relative path to this operation
56 :type path: str
57 :param operation: swagger operation object
58 :type operation: dict
59 :param resolver: Callable that maps operationID to a function
60 :type resolver: resolver.Resolver
61 :param app_produces: list of content types the application can return by default
62 :type app_produces: list
63 :param app_consumes: list of content types the application consumes by default
64 :type app_consumes: list
65 :param path_parameters: Parameters defined in the path level
66 :type path_parameters: list
67 :param app_security: list of security rules the application uses by default
68 :type app_security: list
69 :param security_schemes: `Security Definitions Object
70 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#security-definitions-object>`_
71 :type security_schemes: dict
72 :param definitions: `Definitions Object
73 <https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#definitionsObject>`_
74 :type definitions: dict
75 :param randomize_endpoint: number of random characters to append to operation name
76 :type randomize_endpoint: integer
77 :param uri_parser_class: class to use for uri parsing
78 :type uri_parser_class: AbstractURIParser
79 """
80 uri_parser_class = uri_parser_class or Swagger2URIParser
82 self._router_controller = operation.get("x-swagger-router-controller")
84 super().__init__(
85 method=method,
86 path=path,
87 operation=operation,
88 resolver=resolver,
89 app_security=app_security,
90 security_schemes=security_schemes,
91 randomize_endpoint=randomize_endpoint,
92 uri_parser_class=uri_parser_class,
93 )
95 self._produces = operation.get("produces", app_produces)
96 self._consumes = operation.get("consumes", app_consumes)
98 self.definitions = definitions or {}
100 self._parameters = operation.get("parameters", [])
101 if path_parameters:
102 self._parameters += path_parameters
104 self._responses = operation.get("responses", {})
106 @classmethod
107 def from_spec(cls, spec, *args, path, method, resolver, **kwargs):
108 return cls(
109 method,
110 path,
111 spec.get_operation(path, method),
112 resolver=resolver,
113 path_parameters=spec.get_path_params(path),
114 app_produces=spec.produces,
115 app_consumes=spec.consumes,
116 app_security=spec.security,
117 security_schemes=spec.security_schemes,
118 definitions=spec.definitions,
119 *args,
120 **kwargs,
121 )
123 @property
124 def request_body(self) -> dict:
125 if not hasattr(self, "_request_body"):
126 body_params = []
127 form_params = []
128 for parameter in self.parameters:
129 if parameter["in"] == "body":
130 body_params.append(parameter)
131 elif parameter["in"] == "formData":
132 form_params.append(parameter)
134 if len(body_params) > 1:
135 raise InvalidSpecification(
136 f"{self.method} {self.path}: There can be one 'body' parameter at most"
137 )
139 if body_params and form_params:
140 raise InvalidSpecification(
141 f"{self.method} {self.path}: 'body' and 'formData' parameters are mutually exclusive"
142 )
144 if body_params:
145 self._request_body = self._transform_json(body_params[0])
146 elif form_params:
147 self._request_body = self._transform_form(form_params)
148 else:
149 self._request_body = {}
151 return self._request_body
153 @property
154 def parameters(self):
155 return self._parameters
157 @property
158 def consumes(self):
159 return self._consumes
161 @property
162 def produces(self):
163 return self._produces
165 def get_path_parameter_types(self):
166 types = {}
167 path_parameters = (p for p in self.parameters if p["in"] == "path")
168 for path_defn in path_parameters:
169 if path_defn.get("type") == "string" and path_defn.get("format") == "path":
170 # path is special case for type 'string'
171 path_type = "path"
172 else:
173 path_type = path_defn.get("type")
174 types[path_defn["name"]] = path_type
175 return types
177 def with_definitions(self, schema):
178 if "schema" in schema:
179 schema["schema"]["definitions"] = self.definitions
180 return schema
182 def response_schema(self, status_code=None, content_type=None):
183 response_definition = self.response_definition(status_code, content_type)
184 return self.with_definitions(response_definition.get("schema", {}))
186 def example_response(self, status_code=None, *args, **kwargs):
187 """
188 Returns example response from spec
189 """
190 # simply use the first/lowest status code, this is probably 200 or 201
191 status_code = status_code or sorted(self._responses.keys())[0]
192 examples_path = [str(status_code), "examples"]
193 schema_example_path = [str(status_code), "schema", "example"]
194 schema_path = [str(status_code), "schema"]
196 try:
197 status_code = int(status_code)
198 except ValueError:
199 status_code = 200
201 if status_code == HTTPStatus.NO_CONTENT:
202 return NoContent, status_code
204 try:
205 return (
206 list(deep_get(self._responses, examples_path).values())[0],
207 status_code,
208 )
209 except KeyError:
210 pass
211 try:
212 return (deep_get(self._responses, schema_example_path), status_code)
213 except KeyError:
214 pass
216 try:
217 schema = deep_get(self._responses, schema_path)
218 except KeyError:
219 return ("No example response or response schema defined.", status_code)
221 return (build_example_from_schema(schema), status_code)
223 def body_name(self, content_type: t.Optional[str] = None) -> str:
224 return self.body_definition(content_type).get("name", "body")
226 def body_schema(self, content_type: t.Optional[str] = None) -> dict:
227 """
228 The body schema definition for this operation.
229 """
230 body_definition = self.body_definition(content_type)
231 return self.with_definitions(body_definition).get("schema", {})
233 def body_definition(self, content_type: t.Optional[str] = None) -> dict:
234 """
235 The body complete definition for this operation.
237 **There can be one "body" parameter at most.**
238 """
239 return self.request_body
241 def _transform_json(self, body_parameter: dict) -> dict:
242 """Translate Swagger2 json parameters into OpenAPI 3 jsonschema spec."""
243 nullable = body_parameter.get("x-nullable")
244 if nullable is not None:
245 body_parameter["schema"]["nullable"] = nullable
246 return body_parameter
248 def _transform_form(self, form_parameters: t.List[dict]) -> dict:
249 """Translate Swagger2 form parameters into OpenAPI 3 jsonschema spec."""
250 properties = {}
251 defaults = {}
252 required = []
253 encoding = {}
255 for param in form_parameters:
256 prop = {}
258 if param["type"] == "file":
259 prop.update(
260 {
261 "type": "string",
262 "format": "binary",
263 }
264 )
265 else:
266 prop["type"] = param["type"]
268 format_ = param.get("format")
269 if format_ is not None:
270 prop["format"] = format_
272 default = param.get("default")
273 if default is not None:
274 prop["default"] = default
275 defaults[param["name"]] = default
277 nullable = param.get("x-nullable")
278 if nullable is not None:
279 prop["nullable"] = nullable
281 if param["type"] == "array":
282 prop["items"] = param.get("items", {})
284 collection_format = param.get("collectionFormat", "csv")
285 try:
286 encoding[param["name"]] = COLLECTION_FORMAT_MAPPING[
287 collection_format
288 ]
289 except KeyError:
290 raise InvalidSpecification(
291 f"The collection format ({collection_format}) is not supported by "
292 f"Connexion as it cannot be mapped to OpenAPI 3."
293 )
295 properties[param["name"]] = prop
297 if param.get("required", False):
298 required.append(param["name"])
300 definition: t.Dict[str, t.Any] = {
301 "schema": {
302 "type": "object",
303 "properties": properties,
304 "required": required,
305 }
306 }
307 if defaults:
308 definition["schema"]["default"] = defaults
309 if encoding:
310 definition["encoding"] = encoding
312 return definition