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 ForeignKey)
67 # TODO OneToOneField is gone, so now this should get the users and error if there are multiple?
68 # or we need to move this to user space instead
69 if user := getattr(api_key, "user", None):
70 self.request.user = user
71
72 # Run the regular auth checks which will look for self.request.user
73 super().check_auth()
74
75
76class APIBaseView(View):
77 # Empty by default, so you need to specifically enable the methods you want.
78 allowed_http_methods = []
79
80 form_class: type["BaseForm"] | None = None
81
82 def object_to_dict(self, obj): # Intentionally untyped
83 raise NotImplementedError(
84 f"object_to_dict() is not implemented on {self.__class__.__name__}"
85 )
86
87 def get_form_response(
88 self,
89 ) -> JsonResponse | JsonResponseCreated | JsonResponseBadRequest:
90 if self.form_class is None:
91 raise NotImplementedError(
92 f"form_class is not set on {self.__class__.__name__}"
93 )
94
95 form = self.form_class(**self.get_form_kwargs())
96
97 if form.is_valid():
98 return self.form_valid(form)
99 else:
100 return self.form_invalid(form)
101
102 def get_form_kwargs(self) -> dict[str, Any]:
103 if not self.request.body:
104 raise ResponseException(ResponseBadRequest("No JSON body provided"))
105
106 try:
107 data = json.loads(self.request.body)
108 except json.JSONDecodeError:
109 raise ResponseException(
110 JsonResponseBadRequest({"error": "Unable to parse JSON"})
111 )
112
113 return {
114 "data": data,
115 "files": self.request.FILES,
116 }
117
118 def form_valid(self, form: "BaseForm") -> JsonResponse | JsonResponseCreated:
119 """
120 Used for PUT and PATCH requests.
121 Can check self.request.method if you want different behavior.
122 """
123 object = form.save() # type: ignore
124 data = self.object_to_dict(object)
125
126 if self.request.method == "POST":
127 return JsonResponseCreated(
128 data,
129 json_dumps_params={
130 "sort_keys": True,
131 },
132 )
133 else:
134 return JsonResponse(
135 data,
136 json_dumps_params={
137 "sort_keys": True,
138 },
139 )
140
141 def form_invalid(self, form: "BaseForm") -> JsonResponseBadRequest:
142 return JsonResponseBadRequest(
143 {"message": "Invalid input", "errors": form.errors.get_json_data()},
144 )
145
146
147class APIObjectListView(CsrfExemptViewMixin, APIBaseView):
148 def load_objects(self) -> None:
149 try:
150 self.objects = self.get_objects()
151 except ObjectDoesNotExist:
152 # Custom 404 with no body
153 raise ResponseException(ResponseNotFound())
154
155 if not self.objects:
156 # Also raise 404 if the object is None
157 raise ResponseException(ResponseNotFound())
158
159 def get_objects(self): # Intentionally untyped for subclasses to type
160 raise NotImplementedError(
161 f"get_objects() is not implemented on {self.__class__.__name__}"
162 )
163
164 def get(self) -> JsonResponseList | ResponseNotFound | ResponseBadRequest:
165 self.load_objects()
166 # TODO paginate??
167 data = [self.object_to_dict(obj) for obj in self.objects]
168 return JsonResponseList(data)
169
170 def post(
171 self,
172 ) -> JsonResponseCreated | ResponseNotFound | ResponseBadRequest:
173 self.load_objects()
174 return self.get_form_response() # type: ignore
175
176
177class APIObjectView(CsrfExemptViewMixin, APIBaseView):
178 """Similar to a DetailView but without all of the context and template logic."""
179
180 def load_object(self) -> None:
181 try:
182 self.object = self.get_object()
183 except ObjectDoesNotExist:
184 # Custom 404 with no body
185 raise ResponseException(ResponseNotFound())
186
187 if not self.object:
188 # Also raise 404 if the object is None
189 raise ResponseException(ResponseNotFound())
190
191 def get_object(self): # Intentionally untyped for subclasses to type
192 """
193 Get an instance of an object (typically a model instance).
194
195 Authorization should be done here too.
196 """
197 raise NotImplementedError(
198 f"get_object() is not implemented on {self.__class__.__name__}"
199 )
200
201 def get_form_kwargs(self) -> dict[str, Any]:
202 kwargs = super().get_form_kwargs()
203 kwargs["instance"] = self.object
204 return kwargs
205
206 def get(self) -> JsonResponse | ResponseNotFound | ResponseBadRequest:
207 self.load_object()
208 data = self.object_to_dict(self.object)
209 return JsonResponse(data)
210
211 def put(self) -> JsonResponse | ResponseNotFound | ResponseBadRequest:
212 self.load_object()
213 return self.get_form_response()
214
215 def patch(self) -> JsonResponse | ResponseNotFound | ResponseBadRequest:
216 self.load_object()
217 return self.get_form_response()
218
219 def delete(
220 self,
221 ) -> HttpNoContentResponse | ResponseNotFound | ResponseBadRequest:
222 self.load_object()
223 self.object.delete()
224 return HttpNoContentResponse()