aircox/aircox/viewsets.py

from django.contrib.auth.models import User, Group
from django_filters import rest_framework as drf_filters
from rest_framework import status, viewsets, parsers, permissions
from rest_framework.decorators import action
from rest_framework.response import Response

from filer.models.imagemodels import Image

from . import models, forms, filters, serializers
from .views import BaseAPIView


__all__ = (
    "ImageViewSet",
    "SoundViewSet",
    "TrackROViewSet",
    "UserGroupViewSet",
    "UserSettingsViewSet",
)


class AutocompleteMixin:
    """Based on provided filterset and serializer, add an "autocomplete" action
    to the viewset.

    Url ``GET`` parameters:
        - `field` (many): if provided, only return provided field names
        - filterset's lookups.

    Return a list of values if ``field`` is provided, result of `list()` otherwise.
    """

    @action(name="autocomplete", detail=False)
    def autocomplete(self, request):
        field = request.GET.get("field", None)
        if field:
            queryset = self.filter_queryset(self.get_queryset())
            values = queryset.values_list(field, flat=True).distinct()
            return Response(values[:10])
        return self.list(request)


class ListCommitMixin:
    @action(name="commit", detail=False, methods=["POST"])
    def commit(self, request):
        """
        Data:
            {
                "delete": [pk],
                "update": [{pk, **object}],
                "create": [object_data]
            }

        Return:
            {
                "deleted": [pk],
                "updated": [object],
                "created": [object],
            }
        """
        queryset = self.get_queryset()
        resp = {"deleted": [], "updated": [], "created": []}
        if ids := request.data.get("delete"):
            q = queryset.filter(id__in=ids)
            resp["deleted"] = list(q.values_list("id", flat=True))
            q.delete()

        # TODO: bulk save and update
        if items := request.data.get("update"):
            resp["updated"] = self._commit_save_many(items)

        if items := request.data.get("create"):
            resp["created"] = self._commit_save_many(items)

        return Response(data=resp)

    def _commit_save_many(self, data):
        ser = self.get_serializer(data=data, many=True)
        ser.is_valid(raise_exception=True)

        items = ser.save()
        ser = self.get_serializer(items, many=True)
        return ser.data


class ImageViewSet(viewsets.ModelViewSet):
    parsers = (parsers.MultiPartParser,)
    permissions = (permissions.IsAuthenticatedOrReadOnly,)
    serializer_class = serializers.admin.ImageSerializer
    queryset = Image.objects.all().order_by("-uploaded_at")
    filter_backends = (drf_filters.DjangoFilterBackend,)
    filterset_class = filters.ImageFilterSet

    def create(self, request, **kwargs):
        # FIXME: to be replaced by regular DRF
        form = forms.ImageForm(request.POST, request.FILES)
        if form.is_valid():
            file = form.cleaned_data["file"]
            Image.objects.create(original_filename=file.name, file=file)
            return Response({"status": "ok"})
        return Response({"status": "error", "errors": form.errors})


class SoundViewSet(BaseAPIView, viewsets.ModelViewSet):
    parsers = (parsers.MultiPartParser,)
    permissions = (permissions.IsAuthenticatedOrReadOnly,)
    serializer_class = serializers.SoundSerializer
    queryset = models.Sound.objects.order_by("-pk")
    filter_backends = (drf_filters.DjangoFilterBackend,)
    filterset_class = filters.SoundFilterSet

    def perform_create(self, serializer):
        obj = serializer.save()
        # FIXME: hack to avoid "TYPE_REMOVED" status
        # -> file is saved to fs after object is saved to db
        obj.save()

    def get_queryset(self):
        query = super().get_queryset()
        if not self.request.user.is_authenticated:
            return query.available()
        return query


class TrackROViewSet(AutocompleteMixin, viewsets.ReadOnlyModelViewSet):
    """Track viewset used for auto completion."""

    serializer_class = serializers.admin.TrackSerializer
    permission_classes = (permissions.IsAuthenticated,)
    filterset_class = filters.TrackFilterSet
    queryset = models.Track.objects.all()


class CommentViewSet(viewsets.ModelViewSet):
    serializer_class = serializers.CommentSerializer
    permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
    queryset = models.Comment.objects.all()


# --- admin
class UserViewSet(AutocompleteMixin, viewsets.ModelViewSet):
    serializer_class = serializers.auth.UserSerializer
    permission_classes = (permissions.IsAdminUser,)
    filterset_class = filters.UserFilterSet
    queryset = User.objects.all().distinct().order_by("username")


class GroupViewSet(AutocompleteMixin, viewsets.ModelViewSet):
    serializer_class = serializers.auth.GroupSerializer
    permission_classes = (permissions.IsAdminUser,)
    filterset_class = filters.GroupFilterSet
    queryset = Group.objects.all().distinct().order_by("name")


class UserGroupViewSet(ListCommitMixin, viewsets.ModelViewSet):
    serializer_class = serializers.auth.UserGroupSerializer
    permission_classes = (permissions.IsAdminUser,)
    filterset_class = filters.UserGroupFilterSet
    model = User.groups.through
    queryset = model.objects.all().distinct().order_by("user__username")


class UserSettingsViewSet(viewsets.ViewSet):
    """User's settings specific to aircox.

    Allow only to create and edit user's own settings.
    """

    serializer_class = serializers.UserSettingsSerializer
    permission_classes = (permissions.IsAuthenticated,)

    def get_serializer(self, instance=None, **kwargs):
        return self.serializer_class(instance=instance, context={"user": self.request.user}, **kwargs)

    @action(detail=False, methods=["GET"])
    def retrieve(self, request):
        user = self.request.user
        settings = getattr(user, "aircox_settings", None)
        data = settings and self.get_serializer(settings) or None
        return Response(data)

    @action(detail=False, methods=["POST", "PUT"])
    def update(self, request):
        user = self.request.user
        settings = getattr(user, "aircox_settings", None)
        data = dict(request.data)
        data["user_id"] = self.request.user
        serializer = self.get_serializer(instance=settings, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response({"status": "ok"})
        else:
            return Response(
                {"errors": serializer.errors},
                status=status.HTTP_400_BAD_REQUEST,
            )