field=models.ForeignKey(
editable=False,
on_delete=django.db.models.deletion.CASCADE,
- related_name="pns_recieved",
+ related_name="pms_recieved",
to=settings.AUTH_USER_MODEL,
),
),
User, related_name="pms_sent", on_delete=CASCADE, editable=False
)
recipient = ForeignKey(
- User, related_name="pns_recieved", on_delete=CASCADE, editable=False
+ User, related_name="pms_recieved", on_delete=CASCADE, editable=False
)
text = TextField()
"recipient_id",
"text",
]
+ update_fields = ["text"]
def from_json(self, data, instance=None):
if instance is None:
{% block head %}
{{ block.super }}
<link rel="stylesheet" href="{% static 'channel/styles.css' %}">
- <script src="{% static 'chat/chatutils.js' %}"></script>
+ <script src="{% static 'channel/chatutils.js' %}"></script>
{% endblock head %}
{% block header %}
{% if messages %}
+from django.db.models import Q
from django.views.generic import TemplateView
+from rest.views import ModelRestView, PrivilegeRequiredMixin
+from .serializers import ChannelSerializer, PrivateMessageSerializer
+
class ChannelMainView(TemplateView):
template_name = "channel/main.html"
+
+
+class PrivateMessageRestView(ModelRestView):
+ serializer = PrivateMessageSerializer
+
+ def get_queryset(self):
+ return (
+ super()
+ .get_queryset()
+ .filter(Q(sender=self.request.user.pk) | Q(recipient=self.request.user.pk))
+ )
+
+
+class ChannelRestView(PrivilegeRequiredMixin, ModelRestView):
+ serializer = ChannelSerializer
+
+ def get_queryset(self):
+ return super().get_queryset().filter(users=self.request.user.pk)
--- /dev/null
+(function () {
+ function fade_messages(event) {
+ var li = event.target;
+ if (li.tagName !== "LI" || li.classList.contains("fade-out")) {
+ return;
+ }
+ li.classList.add("fade-out");
+ li.addEventListener(
+ "transitionend",
+ function () {
+ li.remove();
+ },
+ {once: true}
+ );
+ }
+ document.addEventListener(
+ "readystatechange",
+ function () {
+ var sel;
+ if (document.readyState !== "complete") {
+ return;
+ }
+ sel = document.querySelector("ul.messages");
+ if (sel !== null) {
+ sel.addEventListener("click", fade_messages);
+ }
+ }
+ );
+}());
html, body, main {
height: 100%;
}
+
+ul.messages {
+ position: fixed;
+ top: 0;
+ left: 0;
+ width: 100%;
+ padding: 10px;
+ list-style: none;
+}
+
+/*
+DEBUG = 10
+INFO = 20
+SUCCESS = 25
+WARNING = 30
+ERROR = 40
+*/
+
+ul.messages li {
+ color: midnightblue;
+ padding: 12px 20px;
+ margin-bottom: 10px;
+ display: flex;
+ justify-content: space-between;
+ align-items: center;
+ transition: opacity 0.5s ease;
+ opacity: 1;
+ cursor: pointer;
+ user-select: none; /* Prevents text highlighting on quick clicks */
+}
+
+ul.messages li.debug {
+ background-color: lightcyan;
+}
+
+ul.messages li.info {
+ background-color: lightblue;
+}
+
+ul.messages li.success {
+ background-color: lightgreen;
+}
+
+ul.messages li.warning {
+ background-color: orange;
+}
+
+ul.messages li.error {
+ background-color: lightcoral;
+}
+
+ul.messages li.fade-out {
+ transition-delay: 0.5s;
+ opacity: 0;
+ pointer-events: none; /* Prevents double-clicks during the fade */
+}
<meta name="keywords" content="{{ meta_keywords }}">
{% endblock meta %}
<link rel="stylesheet" href="{% static 'chat/styles.css' %}">
+ <script src="{% static 'chat/base.js' %}"></script>
<title>
{% block title %}
{{ title|default:'Untitled' }}
class ModelSerializer:
model: Type[Model]
fields = ["id", "url"]
+ create_fields = None
+ update_fields = None
- def __init__(self, request):
+ def __init__(self, request=None):
self.request = request
def field_to_json(self, field_name, instance):
def to_json(self, instance):
return {key: self.field_to_json(key, instance) for key in self.fields}
- def from_json(self, data, instance=None):
- if instance is None:
- instance = self.model()
+ def get_fields_attr(self):
+ if self.request.method == "POST" and self.create_fields is not None:
+ return "create_fields"
+ elif self.request.method == "PUT" and self.update_fields is not None:
+ return "update_fields"
+ return "fields"
+
+ def from_json(self, data, instance):
m2m = {}
- for field_name, value in data.items():
- if field_name not in self.fields or field_name in ("id", "url"):
+ fields_attr = self.get_fields_attr()
+ for field_name in getattr(self, fields_attr):
+ if field_name not in data or field_name in ("id", "url"):
continue
+ value = data[field_name]
field = self.model._meta.get_field(field_name)
if isinstance(field, DateTimeField):
value = datetime.fromisoformat(value)
m2m[field_name] = value
continue
if isinstance(field, ForeignKey):
- # disallow reassigning FKs on update
- if instance.pk is not None:
+ # when no update_fields are specified, disallow assigning FKs on update
+ if fields_attr == "fields" and instance.pk is not None:
raise ValueError(f"Not allowed to update: {field_name}")
+ # TODO: validate field lists elsewhere
if not field_name.endswith("_id") or not isinstance(value, int):
raise KeyError(f"Use fk_id fields with an id: {field_name}")
setattr(instance, field_name, value)
return instance, m2m
+
+ def save(self, data, instance):
+ instance, m2m = self.from_json(data, instance)
+ instance.save()
+ for field_name, value in m2m.items():
+ getattr(instance, field_name).set(value)
def get_urls(view_class, name, *extra_patterns):
- urlpatterns.extend(
- (
+ for method_map in (
+ {"get": "list", "post": "create"},
+ {"get": "detail", "put": "update", "delete": "delete"},
+ ):
+ urlpatterns.append(
path(
f"{name}/",
- view_class.as_view(method_map={"get": "list", "post": "create"}),
+ view_class.as_view(method_map=method_map),
name=f"api-{name}-list",
- ),
- path(
- f"{name}/<int:id>/",
- view_class.as_view(
- method_map={"get": "detail", "put": "update", "delete": "delete"}
- ),
- name=f"api-{name}-detail",
- ),
- *extra_patterns,
+ )
)
- )
+ urlpatterns.extend(extra_patterns)
import json
from functools import cached_property
from typing import Type
-from urllib.parse import urlunsplit
from django.conf import settings
-from django.contrib.auth import logout
from django.db.models import QuerySet
-from django.http import HttpResponse
-from django.urls import reverse
+from django.http import HttpResponse, HttpResponseNotAllowed
from django.utils.decorators import method_decorator
from django.views import View
from django.views.decorators.csrf import csrf_exempt
@method_decorator(csrf_exempt, name="dispatch")
class ModelRestView(View):
- http_method_names = []
- list_all = False
method_map = {}
-
serializer: Type[ModelSerializer]
def get_queryset(self):
return QuerySet(self.serializer.model).all()
- safe_chars = "/#%[]=:;$&()+,!?*@'~"
+ def get_object(self):
+ return self.get_queryset().get(pk=self.kwargs["id"])
def paginate(self, queryset, to_json):
query = self.request.GET.copy()
if "since" in query:
queryset = queryset.filter(pk__gte=int(query["since"]))
page_size = queryset.count()
- elif not "list_all":
+ elif "list_all" not in query:
page_size = settings.DEFAULT_PAGE_SIZE
queryset = queryset[max(count - page_size, 0) :]
else:
if count > page_size:
query["before"] = queryset[0].pk
prev_url = self.request.build_absolute_uri(
- urlunsplit(
- ("", "", self.request.path, query.urlencode(self.safe_chars), "")
- ),
+ f"{self.request.path}?{query.urlencode('/:@[]')}"
)
else:
prev_url = None
"result": [to_json(instance) for instance in queryset],
}
- def get_object(self):
- return self.get_queryset().get(pk=self.kwargs["id"])
-
@staticmethod
def get_json_dump_kwargs():
if settings.DEBUG: # pragma: no cover
)
def create(self):
- return self.create_or_update(status=201)
+ serializer = self.serializer(self.request)
+ instance = serializer.model()
+ serializer.save(json.load(self.request), instance)
+ return HttpResponse(
+ json.dumps(serializer.to_json(instance)),
+ content_type="application/json",
+ status=201,
+ )
def detail(self):
return HttpResponse(
)
def update(self):
- return self.create_or_update(self.get_object())
-
- def create_or_update(self, *args, **kwargs):
serializer = self.serializer(self.request)
- instance, m2m = serializer.from_json(json.load(self.request), *args)
- instance.save()
- for field_name, value in m2m.items():
- getattr(instance, field_name).set(value)
- if settings.REST_CREATE_UPDATE_RETURN_RESULT:
- return HttpResponse(
- json.dumps(serializer.to_json(instance)),
- content_type="application/json",
- **kwargs,
- )
- return HttpResponse(status=204) # pragma: no cover
+ instance = self.get_object()
+ serializer.save(json.load(self.request), instance)
+ return HttpResponse(
+ json.dumps(serializer.to_josn(instance)),
+ content_type="application/json",
+ )
def delete(self):
self.get_object().delete()
return self.method_map[self.request.method.lower()]
def dispatch(self, request, *args, **kwargs):
- if not request.user.is_authenticated:
- return self.handle_no_permission()
try:
return getattr(self, self.action)()
except Exception as e:
status=500,
)
- def handle_no_permission(self):
- logout(self.request)
- return HttpResponse(
- json.dumps({"Location": reverse("login")}),
- content_type="application/json",
- status=401,
- )
+
+class PrivilegeRequiredMixin:
+ def dispatch(self, request, *args, **kwargs):
+ if request.method not in (permitted_methods := self.get_permitted_methods()):
+ return HttpResponseNotAllowed(permitted_methods)
+ return super().dispatch(request, *args, **kwargs)
+
+ def get_permitted_methods(self):
+ if self.request.user.is_privileged():
+ return ["GET", "PUT", "DELETE"] if "id" in self.kwargs else ["GET", "POST"]
+ return ["GET"]
"date_joined",
"channels",
]
+ update_fields = ["username", "first_name", "last_name"]
def to_json(self, instance):
result = super().to_json(instance)
{% endblock head %}
{% block main %}
<h1>{{ title }}</h1>
- {% if messages %}
+ {% if messages or True %}
<ul class="messages">
{% for message in messages %}
<li class="{{ message.tags|default:'' }}">{{ message }}</li>
from django.views.generic import FormView, RedirectView
from chat.bridge import bridge
-from rest.views import ModelRestView
+from rest.views import ModelRestView, PrivilegeRequiredMixin
from .email import (
send_password_reset_email,
send_register_confirmation_email,
return super().form_valid(form)
-class UserRestView(ModelRestView):
+class UserRestView(PrivilegeRequiredMixin, ModelRestView):
serializer = UserSerializer
def get_queryset(self):
user_id = self.request.user.pk
if self.request.resolver_match.view_name == "api-user-current":
self.kwargs["id"] = user_id
- if (
- self.action in ("create", "delete") and not request.user.is_privileged()
- ) or (
- self.action == "update"
- and not request.user.is_privileged()
- and self.kwargs["id"] != user_id
- ):
- return self.handle_no_permission()
return super().dispatch(request, *args, **kwargs)
+
+ def get_permitted_methods(self):
+ user = self.request.user
+ if not user.is_privileged() and self.kwargs.get("id") == user.pk:
+ return ["GET", "PUT"]
+ return super().get_permitted_methods()
projects / chat / commitdiff