forked from openapi-generators/openapi-python-client
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathopenapi.py
More file actions
284 lines (245 loc) · 11.4 KB
/
openapi.py
File metadata and controls
284 lines (245 loc) · 11.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
from copy import deepcopy
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Tuple, Union
from pydantic import ValidationError
from .. import schema as oai
from .. import utils
from .errors import GeneratorError, ParseError, PropertyError
from .properties import EnumProperty, ModelProperty, Property, Schemas, build_schemas, property_from_data
from .reference import Reference
from .responses import Response, response_from_data
class ParameterLocation(str, Enum):
""" The places Parameters can be put when calling an Endpoint """
QUERY = "query"
PATH = "path"
HEADER = "header"
def import_string_from_reference(reference: Reference, prefix: str = "") -> str:
""" Create a string which is used to import a reference """
return f"from {prefix}.{reference.module_name} import {reference.class_name}"
@dataclass
class EndpointCollection:
""" A bunch of endpoints grouped under a tag that will become a module """
tag: str
endpoints: List["Endpoint"] = field(default_factory=list)
parse_errors: List[ParseError] = field(default_factory=list)
@staticmethod
def from_data(
*, data: Dict[str, oai.PathItem], schemas: Schemas
) -> Tuple[Dict[str, "EndpointCollection"], Schemas]:
""" Parse the openapi paths data to get EndpointCollections by tag """
endpoints_by_tag: Dict[str, EndpointCollection] = {}
methods = ["get", "put", "post", "delete", "options", "head", "patch", "trace"]
for path, path_data in data.items():
for method in methods:
operation: Optional[oai.Operation] = getattr(path_data, method)
if operation is None:
continue
tag = (operation.tags or ["default"])[0]
collection = endpoints_by_tag.setdefault(tag, EndpointCollection(tag=tag))
endpoint, schemas = Endpoint.from_data(
data=operation, path=path, method=method, tag=tag, schemas=schemas
)
if isinstance(endpoint, ParseError):
endpoint.header = (
f"ERROR parsing {method.upper()} {path} within {tag}. Endpoint will not be generated."
)
collection.parse_errors.append(endpoint)
continue
for error in endpoint.errors:
error.header = f"WARNING parsing {method.upper()} {path} within {tag}."
collection.parse_errors.append(error)
collection.endpoints.append(endpoint)
return endpoints_by_tag, schemas
def generate_operation_id(*, path: str, method: str) -> str:
""" Generate an operationId from a path """
clean_path = path.replace("{", "").replace("}", "").replace("/", "_")
if clean_path.startswith("_"):
clean_path = clean_path[1:]
if clean_path.endswith("_"):
clean_path = clean_path[:-1]
return f"{method}_{clean_path}"
@dataclass
class Endpoint:
"""
Describes a single endpoint on the server
"""
path: str
method: str
description: Optional[str]
name: str
requires_security: bool
tag: str
relative_imports: Set[str] = field(default_factory=set)
query_parameters: List[Property] = field(default_factory=list)
path_parameters: List[Property] = field(default_factory=list)
header_parameters: List[Property] = field(default_factory=list)
responses: List[Response] = field(default_factory=list)
form_body_reference: Optional[Reference] = None
json_body: Optional[Property] = None
multipart_body_reference: Optional[Reference] = None
errors: List[ParseError] = field(default_factory=list)
@staticmethod
def parse_request_form_body(body: oai.RequestBody) -> Optional[Reference]:
""" Return form_body_reference """
body_content = body.content
form_body = body_content.get("application/x-www-form-urlencoded")
if form_body is not None and isinstance(form_body.media_type_schema, oai.Reference):
return Reference.from_ref(form_body.media_type_schema.ref)
return None
@staticmethod
def parse_multipart_body(body: oai.RequestBody) -> Optional[Reference]:
""" Return form_body_reference """
body_content = body.content
json_body = body_content.get("multipart/form-data")
if json_body is not None and isinstance(json_body.media_type_schema, oai.Reference):
return Reference.from_ref(json_body.media_type_schema.ref)
return None
@staticmethod
def parse_request_json_body(
*, body: oai.RequestBody, schemas: Schemas, parent_name: str
) -> Tuple[Union[Property, PropertyError, None], Schemas]:
""" Return json_body """
body_content = body.content
json_body = body_content.get("application/json")
if json_body is not None and json_body.media_type_schema is not None:
return property_from_data(
name="json_body",
required=True,
data=json_body.media_type_schema,
schemas=schemas,
parent_name=parent_name,
)
return None, schemas
@staticmethod
def _add_body(
*, endpoint: "Endpoint", data: oai.Operation, schemas: Schemas
) -> Tuple[Union[ParseError, "Endpoint"], Schemas]:
""" Adds form or JSON body to Endpoint if included in data """
endpoint = deepcopy(endpoint)
if data.requestBody is None or isinstance(data.requestBody, oai.Reference):
return endpoint, schemas
endpoint.form_body_reference = Endpoint.parse_request_form_body(data.requestBody)
json_body, schemas = Endpoint.parse_request_json_body(
body=data.requestBody, schemas=schemas, parent_name=endpoint.name
)
if isinstance(json_body, ParseError):
return ParseError(detail=f"cannot parse body of endpoint {endpoint.name}", data=json_body.data), schemas
endpoint.multipart_body_reference = Endpoint.parse_multipart_body(data.requestBody)
if endpoint.form_body_reference:
endpoint.relative_imports.add(
import_string_from_reference(endpoint.form_body_reference, prefix="...models")
)
if endpoint.multipart_body_reference:
endpoint.relative_imports.add(
import_string_from_reference(endpoint.multipart_body_reference, prefix="...models")
)
if json_body is not None:
endpoint.json_body = json_body
endpoint.relative_imports.update(endpoint.json_body.get_imports(prefix="..."))
return endpoint, schemas
@staticmethod
def _add_responses(*, endpoint: "Endpoint", data: oai.Responses, schemas: Schemas) -> Tuple["Endpoint", Schemas]:
endpoint = deepcopy(endpoint)
for code, response_data in data.items():
response, schemas = response_from_data(
status_code=int(code), data=response_data, schemas=schemas, parent_name=endpoint.name
)
if isinstance(response, ParseError):
endpoint.errors.append(
ParseError(
detail=(
f"Cannot parse response for status code {code}, "
f"response will be ommitted from generated client"
),
data=response.data,
)
)
continue
endpoint.relative_imports |= response.prop.get_imports(prefix="...")
endpoint.responses.append(response)
return endpoint, schemas
@staticmethod
def _add_parameters(
*, endpoint: "Endpoint", data: oai.Operation, schemas: Schemas
) -> Tuple[Union["Endpoint", ParseError], Schemas]:
endpoint = deepcopy(endpoint)
if data.parameters is None:
return endpoint, schemas
for param in data.parameters:
if isinstance(param, oai.Reference) or param.param_schema is None:
continue
prop, schemas = property_from_data(
name=param.name,
required=param.required,
data=param.param_schema,
schemas=schemas,
parent_name=endpoint.name,
)
if isinstance(prop, ParseError):
return ParseError(detail=f"cannot parse parameter of endpoint {endpoint.name}", data=prop.data), schemas
endpoint.relative_imports.update(prop.get_imports(prefix="..."))
if param.param_in == ParameterLocation.QUERY:
endpoint.query_parameters.append(prop)
elif param.param_in == ParameterLocation.PATH:
endpoint.path_parameters.append(prop)
elif param.param_in == ParameterLocation.HEADER:
endpoint.header_parameters.append(prop)
else:
return ParseError(data=param, detail="Parameter must be declared in path or query"), schemas
return endpoint, schemas
@staticmethod
def from_data(
*, data: oai.Operation, path: str, method: str, tag: str, schemas: Schemas
) -> Tuple[Union["Endpoint", ParseError], Schemas]:
""" Construct an endpoint from the OpenAPI data """
if data.operationId is None:
name = generate_operation_id(path=path, method=method)
else:
name = data.operationId
endpoint = Endpoint(
path=path,
method=method,
description=utils.remove_string_escapes(data.description) if data.description else "",
name=name,
requires_security=bool(data.security),
tag=tag,
)
result, schemas = Endpoint._add_parameters(endpoint=endpoint, data=data, schemas=schemas)
if isinstance(result, ParseError):
return result, schemas
result, schemas = Endpoint._add_responses(endpoint=result, data=data.responses, schemas=schemas)
result, schemas = Endpoint._add_body(endpoint=result, data=data, schemas=schemas)
return result, schemas
@dataclass
class GeneratorData:
""" All the data needed to generate a client """
title: str
description: Optional[str]
version: str
models: Dict[str, ModelProperty]
errors: List[ParseError]
endpoint_collections_by_tag: Dict[str, EndpointCollection]
enums: Dict[str, EnumProperty]
@staticmethod
def from_dict(d: Dict[str, Dict[str, Any]]) -> Union["GeneratorData", GeneratorError]:
""" Create an OpenAPI from dict """
try:
openapi = oai.OpenAPI.parse_obj(d)
except ValidationError as e:
return GeneratorError(header="Failed to parse OpenAPI document", detail=str(e))
if openapi.components is None or openapi.components.schemas is None:
schemas = Schemas()
else:
schemas = build_schemas(components=openapi.components.schemas)
endpoint_collections_by_tag, schemas = EndpointCollection.from_data(data=openapi.paths, schemas=schemas)
enums = schemas.enums
return GeneratorData(
title=openapi.info.title,
description=openapi.info.description,
version=openapi.info.version,
endpoint_collections_by_tag=endpoint_collections_by_tag,
models=schemas.models,
errors=schemas.errors,
enums=enums,
)