1from http import HTTPStatus
2
3from plain.forms import fields
4from plain.forms.forms import BaseForm
5
6from .utils import merge_data, schema_from_type
7
8
9def response_typed_dict(
10 status_code: int | HTTPStatus | str,
11 return_type,
12 *,
13 description="",
14 component_name="",
15):
16 """
17 A decorator to attach responses to a view method.
18 """
19
20 def decorator(func):
21 # TODO if return_type is a list/tuple,
22 # then use anyOf or oneOf?
23
24 response_schema = {
25 "description": description or HTTPStatus(int(status_code)).phrase,
26 }
27
28 # If we have a return_type, then make it a component and add it
29 # to the response and components
30 if return_type:
31 return_component_name = return_type.__name__
32 response_schema["content"] = {
33 "application/json": {
34 "schema": {"$ref": f"#/components/schemas/{return_component_name}"}
35 }
36 }
37 _component_schema = {
38 "schemas": {
39 return_component_name: schema_from_type(return_type),
40 },
41 }
42 func.openapi_components = merge_data(
43 getattr(func, "openapi_components", {}),
44 _component_schema,
45 )
46
47 if component_name:
48 _schema = {
49 "responses": {
50 str(status_code): {
51 "$ref": f"#/components/responses/{component_name}"
52 }
53 }
54 }
55 func.openapi_components = merge_data(
56 getattr(func, "openapi_components", {}),
57 {
58 "responses": {
59 component_name: response_schema,
60 }
61 },
62 )
63 else:
64 _schema = {"responses": {str(status_code): response_schema}}
65
66 # Add the response schema to the function
67 func.openapi_schema = merge_data(
68 getattr(func, "openapi_schema", {}),
69 _schema,
70 )
71
72 return func
73
74 return decorator
75
76
77def request_form(form_class: BaseForm):
78 """
79 Create OpenAPI parameters from a form class.
80 """
81
82 def decorator(func):
83 field_mappings = {
84 fields.IntegerField: {
85 "type": "integer",
86 },
87 fields.FloatField: {
88 "type": "number",
89 },
90 fields.DateTimeField: {
91 "type": "string",
92 "format": "date-time",
93 },
94 fields.DateField: {
95 "type": "string",
96 "format": "date",
97 },
98 fields.TimeField: {
99 "type": "string",
100 "format": "time",
101 },
102 fields.EmailField: {
103 "type": "string",
104 "format": "email",
105 },
106 fields.URLField: {
107 "type": "string",
108 "format": "uri",
109 },
110 fields.UUIDField: {
111 "type": "string",
112 "format": "uuid",
113 },
114 fields.DecimalField: {
115 "type": "number",
116 },
117 # fields.FileField: {
118 # "type": "string",
119 # "format": "binary",
120 # },
121 fields.ImageField: {
122 "type": "string",
123 "format": "binary",
124 },
125 fields.BooleanField: {
126 "type": "boolean",
127 },
128 fields.NullBooleanField: {
129 "type": "boolean",
130 "nullable": True,
131 },
132 fields.CharField: {
133 "type": "string",
134 },
135 fields.EmailField: {
136 "type": "string",
137 "format": "email",
138 },
139 }
140 _schema = {
141 "requestBody": {
142 "content": {
143 "application/json": {
144 "schema": {
145 "type": "object",
146 "properties": {},
147 }
148 }
149 # could add application/x-www-form-urlencoded?
150 }
151 }
152 }
153
154 required_fields = []
155
156 for field_name, field in form_class.base_fields.items():
157 field_schema = field_mappings[field.__class__].copy()
158 _schema["requestBody"]["content"]["application/json"]["schema"][
159 "properties"
160 ][field_name] = field_schema
161
162 if field.required:
163 required_fields.append(field_name)
164
165 # TODO add description to the schema
166 # TODO add example to the schema
167 # TODO add default to the schema
168
169 if required_fields:
170 _schema["requestBody"]["content"]["application/json"]["schema"][
171 "required"
172 ] = required_fields
173 # The body is required if any field is
174 _schema["requestBody"]["required"] = True
175
176 func.openapi_schema = merge_data(
177 getattr(func, "openapi_schema", {}),
178 _schema,
179 )
180
181 return func
182
183 return decorator
184
185
186def schema(data):
187 """
188 A decorator to attach raw OpenAPI schema to a router, view, or view method.
189 """
190
191 def decorator(func):
192 func.openapi_schema = merge_data(
193 getattr(func, "openapi_schema", {}),
194 data,
195 )
196 return func
197
198 return decorator