140 lines
4.7 KiB
Python

# -*- encoding: utf-8 -*-
import logging
from django.db.utils import IntegrityError
from django.core.exceptions import ValidationError
from django_filters import rest_framework as rest_filters
from rest_framework import status, permissions, filters, generics
from rest_framework.response import Response
from rest_framework.pagination import PageNumberPagination
from rest_framework.authentication import SessionAuthentication, TokenAuthentication
from rest_framework.parsers import MultiPartParser, FormParser
from apps.home.models import Subscription, TrackedContent
from .serializers import (
SubscriptionSerializer,
TrackedContentSerializer
)
log = logging.getLogger(__name__)
class DefaultPagination(PageNumberPagination):
"""Default class for pagination in API views."""
page_size = 10
page_size_query_param = "page_size"
max_page_size = 25
# =================================================================================================
# Subscription Views
class Subscription_ListView(generics.ListCreateAPIView):
"""
View to provide a list of Subscription model instances.
Can also be used to create a new instance.
Supports: GET, POST
"""
authentication_classes = [SessionAuthentication, TokenAuthentication]
permission_classes = [permissions.IsAuthenticated]
pagination_class = DefaultPagination
serializer_class = SubscriptionSerializer
queryset = Subscription.objects.all().order_by("-creation_datetime")
filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter]
filterset_fields = ["uuid", "name", "rss_url", "server", "targets", "creation_datetime", "extra_notes", "active"]
search_fields = ["name", "extra_notes"]
ordering_fields = ["creation_datetime"]
def post(self, request):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
self.perform_create(serializer)
except IntegrityError:
return Response(
{"detail": "Duplicate or limit reached"},
status=status.HTTP_409_CONFLICT,
exception=True
)
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
class Subscription_DetailView(generics.RetrieveUpdateDestroyAPIView):
"""
View to provide details on a particular Subscription model instances.
Supports: GET, DELETE
"""
authentication_classes = [SessionAuthentication, TokenAuthentication]
permission_classes = [permissions.IsAuthenticated]
parser_classes = [MultiPartParser, FormParser]
serializer_class = SubscriptionSerializer
queryset = Subscription.objects.all().order_by("-creation_datetime")
# =================================================================================================
# Tracked Content Views
class TrackedContent_ListView(generics.ListCreateAPIView):
"""
View to provide a list of TrackedContent model instances.
Can also be used to create a new instance.
Supports: GET, POST
"""
authentication_classes = [SessionAuthentication, TokenAuthentication]
permission_classes = [permissions.IsAuthenticated]
pagination_class = DefaultPagination
serializer_class = TrackedContentSerializer
queryset = TrackedContent.objects.all().order_by("-creation_datetime")
filter_backends = [filters.SearchFilter, rest_filters.DjangoFilterBackend, filters.OrderingFilter]
filterset_fields = ["uuid", "subscription", "content_url", "creation_datetime"]
search_fields = ["name"]
ordering_fields = ["creation_datetime"]
def post(self, request):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
self.perform_create(serializer)
except IntegrityError:
return Response(
{"detail": "Tracked content must be unique"},
status=status.HTTP_409_CONFLICT,
exception=True
)
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
class TrackedContent_DetailView(generics.RetrieveDestroyAPIView):
"""
View to provide details on a particular TrackedContent model instances.
Supports: GET, DELETE
"""
authentication_classes = [SessionAuthentication, TokenAuthentication]
permission_classes = [permissions.IsAuthenticated]
parser_classes = [MultiPartParser, FormParser]
serializer_class = TrackedContentSerializer
queryset = TrackedContent.objects.all().order_by("-creation_datetime")