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