forked from openapi-generators/openapi-python-client
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathresponses.py
More file actions
110 lines (91 loc) · 3.24 KB
/
responses.py
File metadata and controls
110 lines (91 loc) · 3.24 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
__all__ = ["Response", "response_from_data"]
from http import HTTPStatus
from typing import Optional, Tuple, Union
import attr
from .. import Config
from .. import schema as oai
from ..utils import PythonIdentifier
from .errors import ParseError, PropertyError
from .properties import AnyProperty, Property, Schemas, property_from_data
@attr.s(auto_attribs=True, frozen=True)
class Response:
"""Describes a single response for an endpoint"""
status_code: HTTPStatus
prop: Property
source: str
def _source_by_content_type(content_type: str) -> Optional[str]:
known_content_types = {
"application/json": "response.json()",
"application/octet-stream": "response.content",
"text/html": "response.text",
}
source = known_content_types.get(content_type)
if source is None and content_type.endswith("+json"):
# Implements https://www.rfc-editor.org/rfc/rfc6838#section-4.2.8 for the +json suffix
source = "response.json()"
return source
def empty_response(
*, status_code: HTTPStatus, response_name: str, config: Config, description: Optional[str]
) -> Response:
"""Return an untyped response, for when no response type is defined"""
return Response(
status_code=status_code,
prop=AnyProperty(
name=response_name,
default=None,
nullable=False,
required=True,
python_name=PythonIdentifier(value=response_name, prefix=config.field_prefix),
description=description,
example=None,
),
source="None",
)
def response_from_data(
*,
status_code: HTTPStatus,
data: Union[oai.Response, oai.Reference],
schemas: Schemas,
parent_name: str,
config: Config,
) -> Tuple[Union[Response, ParseError], Schemas]:
"""Generate a Response from the OpenAPI dictionary representation of it"""
response_name = f"response_{status_code}"
if isinstance(data, oai.Reference):
return (
empty_response(status_code=status_code, response_name=response_name, config=config, description=None),
schemas,
)
content = data.content
if not content:
return (
empty_response(
status_code=status_code, response_name=response_name, config=config, description=data.description
),
schemas,
)
for content_type, media_type in content.items():
source = _source_by_content_type(content_type)
if source is not None:
schema_data = media_type.media_type_schema
break
else:
return ParseError(data=data, detail=f"Unsupported content_type {content}"), schemas
if schema_data is None:
return (
empty_response(
status_code=status_code, response_name=response_name, config=config, description=data.description
),
schemas,
)
prop, schemas = property_from_data(
name=response_name,
required=True,
data=schema_data,
schemas=schemas,
parent_name=parent_name,
config=config,
)
if isinstance(prop, PropertyError):
return prop, schemas
return Response(status_code=status_code, prop=prop, source=source), schemas