1from collections.abc import Mapping
2from functools import cached_property
3from http.client import responses as http_status_phrases
4from typing import Any, ClassVar, cast
5
6from plain.exceptions import ValidationError
7from plain.forms.exceptions import FormFieldMissingError
8from plain.http import (
9 HTTPException,
10 JsonResponse,
11 NotFoundError404,
12 Response,
13 status_for_exception,
14 status_omits_body,
15)
16from plain.utils import timezone
17from plain.utils.cache import patch_cache_control
18from plain.views.base import View
19from plain.views.exceptions import ResponseException
20
21from . import openapi
22from .schemas import ErrorSchema, FieldError
23
24# Allow plain.api to be used without plain.postgres
25try:
26 from .models import APIKey
27except ImportError:
28 APIKey: Any = None
29
30__all__ = [
31 "APIKeyView",
32 "APIResult",
33 "APIView",
34 "JsonNotFoundView",
35]
36
37# `Mapping[str, Any]` (vs `dict[str, Any]`) lets `def get(self) -> MyTypedDict:`
38# satisfy Liskov against the base view — TypedDicts aren't `dict` per PEP 589.
39type APIResult = (
40 Response
41 | None
42 | Mapping[str, Any]
43 | list[Any]
44 | tuple[int, dict[str, Any] | list[Any]]
45)
46
47
48def _error_response(
49 *,
50 error_id: str,
51 message: str,
52 status_code: int,
53 errors: list[FieldError] | None = None,
54) -> JsonResponse:
55 body: ErrorSchema = {"id": error_id, "message": message}
56 if errors is not None:
57 body["errors"] = errors
58 return JsonResponse(body, status_code=status_code)
59
60
61def _validation_field_errors(exc: ValidationError) -> list[FieldError] | None:
62 """Flatten a field-dict ValidationError into a list of `{field, message}`.
63
64 Returns None for string- or list-shaped errors that have no field context.
65 """
66 if not hasattr(exc, "error_dict"):
67 return None
68 return [
69 {"field": field, "message": message}
70 for field, messages in exc
71 for message in messages
72 ]
73
74
75# Snake-case ids are part of the public API surface — client libs key off them.
76_STATUS_ERROR_IDS = {
77 400: "bad_request",
78 401: "unauthorized",
79 403: "permission_denied",
80 404: "not_found",
81 405: "method_not_allowed",
82 409: "conflict",
83 415: "unsupported_media_type",
84 429: "rate_limited",
85}
86
87
88# @openapi.response_typed_dict(400, ErrorSchema)
89# @openapi.response_typed_dict(401, ErrorSchema)
90class APIKeyView(View[APIResult]):
91 api_key_required = True
92
93 # Picked up by the OpenAPI generator: each entry is added to
94 # `components.securitySchemes` and required on every operation served by
95 # this view. Subclasses can override to declare a different scheme.
96 openapi_security_schemes: ClassVar[dict[str, dict[str, Any]]] = {
97 "BearerAuth": {
98 "type": "http",
99 "scheme": "bearer",
100 }
101 }
102
103 @cached_property
104 def api_key(self) -> Any:
105 return self.get_api_key()
106
107 def before_request(self) -> None:
108 if self.api_key:
109 self.use_api_key()
110 elif self.api_key_required:
111 raise ResponseException(
112 _error_response(
113 error_id="api_key_required",
114 message="API key required",
115 status_code=401,
116 )
117 )
118
119 def after_response(self, response: Response) -> Response:
120 response = super().after_response(response)
121 # Make sure it at least has private as a default
122 patch_cache_control(response, private=True)
123 return response
124
125 def use_api_key(self) -> None:
126 """
127 Use the API key for this request.
128
129 Override this to perform other actions with a valid API key.
130 """
131 self.api_key.last_used_at = timezone.now()
132 self.api_key.update(fields=["last_used_at"])
133
134 def get_api_key(self) -> Any:
135 """
136 Get the API key from the request.
137
138 Override this if you want to use a different input method.
139 """
140 if "Authorization" in self.request.headers:
141 header_value = self.request.headers["Authorization"]
142 try:
143 header_token = header_value.split("Bearer ")[1]
144 except IndexError:
145 raise ResponseException(
146 _error_response(
147 error_id="invalid_authorization_header",
148 message="Invalid Authorization header",
149 status_code=400,
150 )
151 )
152
153 try:
154 api_key = APIKey.query.get(token=header_token)
155 except APIKey.DoesNotExist:
156 raise ResponseException(
157 _error_response(
158 error_id="invalid_api_token",
159 message="Invalid API token",
160 status_code=400,
161 )
162 )
163
164 if api_key.is_expired():
165 raise ResponseException(
166 _error_response(
167 error_id="api_token_expired",
168 message="API token has expired",
169 status_code=400,
170 )
171 )
172
173 return api_key
174
175
176@openapi.response_typed_dict(400, ErrorSchema, component_name="BadRequest")
177@openapi.response_typed_dict(401, ErrorSchema, component_name="Unauthorized")
178@openapi.response_typed_dict(403, ErrorSchema, component_name="Forbidden")
179@openapi.response_typed_dict(404, ErrorSchema, component_name="NotFound")
180@openapi.response_typed_dict(
181 "5XX", ErrorSchema, description="Unexpected Error", component_name="ServerError"
182)
183class APIView(View[APIResult]):
184 def convert_result_to_response(self, result: APIResult) -> Response:
185 if isinstance(result, Response):
186 return result
187
188 if result is None:
189 raise NotFoundError404
190
191 status_code = 200
192
193 if isinstance(result, tuple):
194 if len(result) != 2:
195 raise ValueError(
196 "Tuple response must be of length 2 (status_code, data)"
197 )
198 status_code, result = cast(tuple[int, dict[str, Any] | list[Any]], result)
199 if status_code is None:
200 # Old contract: None meant JsonResponse's default 200.
201 status_code = 200
202
203 if isinstance(result, dict | list):
204 # The isinstance guard keeps a non-int status (str, etc.) out
205 # of the predicate — Response construction below remains the
206 # single validator for type and range.
207 if isinstance(status_code, int) and status_omits_body(status_code):
208 # A bodiless status can't carry JSON — `return 204, {}`
209 # sends an empty 204; anything else is a contradiction.
210 if result:
211 raise ValueError(
212 f"A {status_code} response cannot include data — "
213 f"return a 200 with the data, or just Response(status_code={status_code})."
214 )
215 # No Content-Type either — there is no representation to
216 # describe (and nothing for version transforms to touch).
217 return Response(status_code=status_code)
218 return JsonResponse(result, status_code=status_code)
219
220 raise TypeError(f"Unexpected APIView return type: {type(result).__name__}")
221
222 def handle_exception(self, exc: Exception) -> Response:
223 if isinstance(exc, ValidationError):
224 errors = _validation_field_errors(exc)
225 if errors is not None:
226 message = "Validation error"
227 else:
228 detail = "; ".join(exc.messages) if exc.messages else str(exc)
229 message = f"Validation error: {detail}"
230 return _error_response(
231 error_id="validation_error",
232 message=message,
233 status_code=400,
234 errors=errors,
235 )
236 if isinstance(exc, FormFieldMissingError):
237 return _error_response(
238 error_id="missing_field",
239 message=f"Missing field: {exc.field_name}",
240 status_code=400,
241 )
242 if isinstance(exc, HTTPException):
243 # Clamped: a poked-in out-of-range status must not crash the
244 # error renderer (subclass definitions are validated already).
245 status_code = status_for_exception(exc)
246 error_id = _STATUS_ERROR_IDS.get(status_code, "http_error")
247 return _error_response(
248 error_id=error_id,
249 message=str(exc) or http_status_phrases.get(status_code, "HTTP error"),
250 status_code=status_code,
251 )
252 return _error_response(
253 error_id="server_error",
254 message="Internal server error",
255 status_code=500,
256 )
257
258
259class JsonNotFoundView(APIView):
260 """Catch-all view that always returns a JSON 404.
261
262 Mount as a regex catch-all at the end of an API router so unmatched
263 paths under your API prefix return a JSON `ErrorSchema` body instead of
264 the framework's HTML 404 page.
265 """
266
267 def before_request(self) -> None:
268 raise NotFoundError404