1import datetime
2import json
3from typing import TYPE_CHECKING, Any
4
5from plain.auth.views import AuthViewMixin
6from plain.exceptions import ObjectDoesNotExist
7from plain.views.base import View
8from plain.views.csrf import CsrfExemptViewMixin
9from plain.views.exceptions import ResponseException
10
11from .responses import (
12 HttpNoContentResponse,
13 JsonResponse,
14 JsonResponseBadRequest,
15 JsonResponseCreated,
16 JsonResponseList,
17 ResponseBadRequest,
18 ResponseNotFound,
19)
20
21if TYPE_CHECKING:
22 from plain.forms import BaseForm
23
24from .models import APIKey
25
26
27class APIAuthViewMixin(AuthViewMixin):
28 # Disable login redirects
29 login_url = None
30
31 def get_api_key(self) -> APIKey | None:
32 if "Authorization" in self.request.headers:
33 header_value = self.request.headers["Authorization"]
34 try:
35 header_token = header_value.split("Bearer ")[1]
36 except IndexError:
37 raise ResponseException(
38 ResponseBadRequest("Invalid Authorization header")
39 )
40
41 try:
42 api_key = APIKey.objects.get(token=header_token)
43 except APIKey.DoesNotExist:
44 raise ResponseException(ResponseBadRequest("Invalid API token"))
45
46 if api_key.expires_at and api_key.expires_at < datetime.datetime.now():
47 raise ResponseException(ResponseBadRequest("API token has expired"))
48
49 return api_key
50
51 def check_auth(self) -> None:
52 if not hasattr(self, "request"):
53 raise AttributeError(
54 "APIAuthViewMixin requires the request attribute to be set."
55 )
56
57 # If the user is already known, exit early
58 if self.request.user:
59 super().check_auth()
60 return
61
62 if api_key := self.get_api_key():
63 # Put the api_key on the request so we can access it
64 self.request.api_key = api_key
65
66 # Set the user if api_key has that attribute (typically from a OneToOneField)
67 if user := getattr(api_key, "user", None):
68 self.request.user = user
69
70 # Run the regular auth checks which will look for self.request.user
71 super().check_auth()
72
73
74class APIBaseView(View):
75 form_class: type["BaseForm"] | None = None
76
77 def object_to_dict(self, obj): # Intentionally untyped
78 raise NotImplementedError(
79 f"object_to_dict() is not implemented on {self.__class__.__name__}"
80 )
81
82 def get_form_response(
83 self,
84 ) -> JsonResponse | JsonResponseCreated | JsonResponseBadRequest:
85 if self.form_class is None:
86 raise NotImplementedError(
87 f"form_class is not set on {self.__class__.__name__}"
88 )
89
90 form = self.form_class(**self.get_form_kwargs())
91
92 if form.is_valid():
93 return self.form_valid(form)
94 else:
95 return self.form_invalid(form)
96
97 def get_form_kwargs(self) -> dict[str, Any]:
98 if not self.request.body:
99 raise ResponseException(ResponseBadRequest("No JSON body provided"))
100
101 try:
102 data = json.loads(self.request.body)
103 except json.JSONDecodeError:
104 raise ResponseException(
105 JsonResponseBadRequest({"error": "Unable to parse JSON"})
106 )
107
108 return {
109 "data": data,
110 "files": self.request.FILES,
111 }
112
113 def form_valid(self, form: "BaseForm") -> JsonResponse | JsonResponseCreated:
114 """
115 Used for PUT and PATCH requests.
116 Can check self.request.method if you want different behavior.
117 """
118 object = form.save() # type: ignore
119 data = self.object_to_dict(object)
120
121 if self.request.method == "POST":
122 return JsonResponseCreated(
123 data,
124 json_dumps_params={
125 "sort_keys": True,
126 },
127 )
128 else:
129 return JsonResponse(
130 data,
131 json_dumps_params={
132 "sort_keys": True,
133 },
134 )
135
136 def form_invalid(self, form: "BaseForm") -> JsonResponseBadRequest:
137 return JsonResponseBadRequest(
138 {"message": "Invalid input", "errors": form.errors.get_json_data()},
139 )
140
141
142class APIObjectListView(CsrfExemptViewMixin, APIBaseView):
143 def load_objects(self) -> None:
144 try:
145 self.objects = self.get_objects()
146 except ObjectDoesNotExist:
147 # Custom 404 with no body
148 raise ResponseException(ResponseNotFound())
149
150 if not self.objects:
151 # Also raise 404 if the object is None
152 raise ResponseException(ResponseNotFound())
153
154 def get_objects(self): # Intentionally untyped for subclasses to type
155 raise NotImplementedError(
156 f"get_objects() is not implemented on {self.__class__.__name__}"
157 )
158
159 def get(self) -> JsonResponseList | ResponseNotFound | ResponseBadRequest:
160 self.load_objects()
161 # TODO paginate??
162 data = [self.object_to_dict(obj) for obj in self.objects]
163 return JsonResponseList(data)
164
165 def post(
166 self,
167 ) -> JsonResponseCreated | ResponseNotFound | ResponseBadRequest:
168 self.load_objects()
169 return self.get_form_response() # type: ignore
170
171
172class APIObjectView(CsrfExemptViewMixin, APIBaseView):
173 """Similar to a DetailView but without all of the context and template logic."""
174
175 def load_object(self) -> None:
176 try:
177 self.object = self.get_object()
178 except ObjectDoesNotExist:
179 # Custom 404 with no body
180 raise ResponseException(ResponseNotFound())
181
182 if not self.object:
183 # Also raise 404 if the object is None
184 raise ResponseException(ResponseNotFound())
185
186 def get_object(self): # Intentionally untyped for subclasses to type
187 """
188 Get an instance of an object (typically a model instance).
189
190 Authorization should be done here too.
191 """
192 raise NotImplementedError(
193 f"get_object() is not implemented on {self.__class__.__name__}"
194 )
195
196 def get_form_kwargs(self) -> dict[str, Any]:
197 kwargs = super().get_form_kwargs()
198 kwargs["instance"] = self.object
199 return kwargs
200
201 def get(self) -> JsonResponse | ResponseNotFound | ResponseBadRequest:
202 self.load_object()
203 data = self.object_to_dict(self.object)
204 return JsonResponse(data)
205
206 def put(self) -> JsonResponse | ResponseNotFound | ResponseBadRequest:
207 self.load_object()
208 return self.get_form_response()
209
210 def patch(self) -> JsonResponse | ResponseNotFound | ResponseBadRequest:
211 self.load_object()
212 return self.get_form_response()
213
214 def delete(
215 self,
216 ) -> HttpNoContentResponse | ResponseNotFound | ResponseBadRequest:
217 self.load_object()
218 self.object.delete()
219 return HttpNoContentResponse()