1from collections.abc import Callable
  2from http import HTTPStatus
  3from typing import Any, TypeVar
  4
  5from plain.forms import fields
  6from plain.forms.forms import BaseForm
  7
  8from .utils import merge_data, schema_from_type
  9
 10F = TypeVar("F", bound=Callable[..., Any])
 11
 12
 13def response_typed_dict(
 14    status_code: int | HTTPStatus | str,
 15    return_type: Any,
 16    *,
 17    description: str = "",
 18    component_name: str = "",
 19) -> Callable[[F], F]:
 20    """
 21    A decorator to attach responses to a view method.
 22    """
 23
 24    def decorator(func: F) -> F:
 25        # TODO if return_type is a list/tuple,
 26        # then use anyOf or oneOf?
 27
 28        response_schema = {
 29            "description": description or HTTPStatus(int(status_code)).phrase,
 30        }
 31
 32        # If we have a return_type, then make it a component and add it
 33        # to the response and components
 34        if return_type:
 35            return_component_name = return_type.__name__
 36            response_schema["content"] = {
 37                "application/json": {
 38                    "schema": {"$ref": f"#/components/schemas/{return_component_name}"}
 39                }
 40            }
 41            _component_schema = {
 42                "schemas": {
 43                    return_component_name: schema_from_type(return_type),
 44                },
 45            }
 46            func.openapi_components = merge_data(  # type: ignore[attr-defined]
 47                getattr(func, "openapi_components", {}),
 48                _component_schema,
 49            )
 50
 51        if component_name:
 52            _schema = {
 53                "responses": {
 54                    str(status_code): {
 55                        "$ref": f"#/components/responses/{component_name}"
 56                    }
 57                }
 58            }
 59            func.openapi_components = merge_data(  # type: ignore[attr-defined]
 60                getattr(func, "openapi_components", {}),
 61                {
 62                    "responses": {
 63                        component_name: response_schema,
 64                    }
 65                },
 66            )
 67        else:
 68            _schema = {"responses": {str(status_code): response_schema}}
 69
 70        # Add the response schema to the function
 71        func.openapi_schema = merge_data(  # type: ignore[attr-defined]
 72            getattr(func, "openapi_schema", {}),
 73            _schema,
 74        )
 75
 76        return func
 77
 78    return decorator
 79
 80
 81def request_form(form_class: type[BaseForm]) -> Callable[[F], F]:
 82    """
 83    Create OpenAPI parameters from a form class.
 84    """
 85
 86    def decorator(func: F) -> F:
 87        field_mappings = {
 88            fields.IntegerField: {
 89                "type": "integer",
 90            },
 91            fields.FloatField: {
 92                "type": "number",
 93            },
 94            fields.DateTimeField: {
 95                "type": "string",
 96                "format": "date-time",
 97            },
 98            fields.DateField: {
 99                "type": "string",
100                "format": "date",
101            },
102            fields.TimeField: {
103                "type": "string",
104                "format": "time",
105            },
106            fields.EmailField: {
107                "type": "string",
108                "format": "email",
109            },
110            fields.URLField: {
111                "type": "string",
112                "format": "uri",
113            },
114            fields.UUIDField: {
115                "type": "string",
116                "format": "uuid",
117            },
118            fields.DecimalField: {
119                "type": "number",
120            },
121            # fields.FileField: {
122            #     "type": "string",
123            #     "format": "binary",
124            # },
125            fields.ImageField: {
126                "type": "string",
127                "format": "binary",
128            },
129            fields.BooleanField: {
130                "type": "boolean",
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():  # type: ignore[attr-defined]
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(  # type: ignore[attr-defined]
177            getattr(func, "openapi_schema", {}),
178            _schema,
179        )
180
181        return func
182
183    return decorator
184
185
186def schema(data: dict[str, Any]) -> Callable[[F], F]:
187    """
188    A decorator to attach raw OpenAPI schema to a router, view, or view method.
189    """
190
191    def decorator(func: F) -> F:
192        func.openapi_schema = merge_data(  # type: ignore[attr-defined]
193            getattr(func, "openapi_schema", {}),
194            data,
195        )
196        return func
197
198    return decorator