-
Ernesto Rico Schmidt authoredErnesto Rico Schmidt authored
views.py 34.87 KiB
#
# steering, Programme/schedule management for AURA
#
# Copyright (C) 2011-2017, 2020, Ernesto Rico Schmidt
# Copyright (C) 2017-2019, Ingo Leindecker
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
# details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import logging
from datetime import date, datetime, time, timedelta
from textwrap import dedent
from django_filters.rest_framework import DjangoFilterBackend
from djangorestframework_camel_case.util import camelize
from drf_spectacular.utils import (
OpenApiParameter,
OpenApiResponse,
OpenApiTypes,
extend_schema,
extend_schema_view,
)
from rest_framework import decorators
from rest_framework import filters as drf_filters
from rest_framework import mixins, permissions, status, viewsets
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.response import Response
from django.conf import settings
from django.contrib.auth.models import User
from django.db import IntegrityError
from django.http import HttpResponseRedirect, JsonResponse
from django.shortcuts import get_object_or_404
from django.utils import timezone
from program import filters
from program.models import (
Category,
FundingCategory,
Host,
Image,
Language,
License,
LinkType,
MusicFocus,
Note,
RadioSettings,
RRule,
Schedule,
ScheduleConflictError,
Show,
TimeSlot,
Topic,
Type,
)
from program.serializers import (
CategorySerializer,
ErrorSerializer,
FundingCategorySerializer,
HostSerializer,
ImageRenderSerializer,
ImageSerializer,
LanguageSerializer,
LicenseSerializer,
LinkTypeSerializer,
MusicFocusSerializer,
NoteSerializer,
RadioSettingsSerializer,
RRuleSerializer,
ScheduleConflictResponseSerializer,
ScheduleCreateUpdateRequestSerializer,
ScheduleDryRunResponseSerializer,
ScheduleResponseSerializer,
ScheduleSerializer,
ShowSerializer,
TimeSlotSerializer,
TopicSerializer,
TypeSerializer,
UserSerializer,
)
from program.services import (
get_timerange_timeslot_entries,
make_schedule_entry,
resolve_conflicts,
)
from program.utils import get_values, parse_date
logger = logging.getLogger(__name__)
@extend_schema_view(
list=extend_schema(
summary="List schedule for a specific date.",
description=(
"Returns a list of the schedule for a specific date."
"Expects parameters `year` (int), `month` (int), and `day` (int) as url components."
"e.g. /program/2024/01/31/"
),
),
)
class APIDayScheduleViewSet(
mixins.ListModelMixin,
viewsets.GenericViewSet,
):
queryset = TimeSlot.objects.all()
serializer_class = TimeSlotSerializer
def list(self, request, year=None, month=None, day=None):
# datetime.combine returns a timezone naive datetime object
if year is None and month is None and day is None:
start = timezone.make_aware(datetime.combine(timezone.now(), time(0, 0)))
else:
start = timezone.make_aware(datetime.combine(date(year, month, day), time(0, 0)))
end = start + timedelta(hours=24)
include_virtual = request.GET.get("include_virtual") == "true"
schedule = [
make_schedule_entry(timeslot_entry=timeslot_entry)
for timeslot_entry in get_timerange_timeslot_entries(start, end, include_virtual)
]
return JsonResponse(camelize(schedule), safe=False)
@extend_schema_view(
list=extend_schema(
summary="List scheduled playout.",
description=(
"Returns a list of the scheduled playout. "
"The schedule will include virtual timeslots to fill unscheduled gaps if requested."
),
),
)
class APIPlayoutViewSet(
mixins.ListModelMixin,
viewsets.GenericViewSet,
):
queryset = TimeSlot.objects.all()
serializer_class = TimeSlotSerializer
filter_backends = [DjangoFilterBackend]
filterset_class = filters.PlayoutFilterSet
def list(self, request, *args, **kwargs):
"""
Return a JSON representation of the scheduled playout.
Called by
- engine (playout) to retrieve timeslots within a given timerange
- internal calendar to retrieve all timeslots for a week
"""
# datetime.now and datetime.combine return timezone naive datetime objects
if request.GET.get("start") is None:
schedule_start = timezone.make_aware(datetime.combine(datetime.now(), time(0, 0)))
else:
schedule_start = timezone.make_aware(
datetime.combine(parse_date(request.GET.get("start")), time(0, 0))
)
if request.GET.get("end") is None:
schedule_end = schedule_start + timedelta(days=7)
else:
schedule_end = timezone.make_aware(
datetime.combine(
parse_date(request.GET.get("end")) + timedelta(days=1), time(0, 0)
)
)
include_virtual = request.GET.get("include_virtual") == "true"
playout = get_timerange_timeslot_entries(schedule_start, schedule_end, include_virtual)
return JsonResponse(camelize(playout), safe=False)
@extend_schema_view(
create=extend_schema(summary="Create a new user."),
retrieve=extend_schema(
summary="Retrieve a single user.",
description="Non-admin users may only retrieve their own user record.",
),
update=extend_schema(
summary="Update an existing user.",
description="Non-admin users may only update their own user record.",
),
partial_update=extend_schema(
summary="Partially update an existing user.",
description="Non-admin users may only update their own user record.",
),
list=extend_schema(
summary="List all users.",
description=(
"The returned list of records will only contain a single record "
"for non-admin users which is their own user account."
),
),
)
class APIUserViewSet(
mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
mixins.ListModelMixin,
viewsets.GenericViewSet,
):
serializer_class = UserSerializer
filter_backends = [drf_filters.SearchFilter]
search_fields = ["username", "first_name", "last_name", "email"]
def get_queryset(self):
"""The queryset is empty if the user is not authenticated, contains all the users if the
method is safe or the requesting user is a superuser, otherwise it only contains the
requesting user."""
user = self.request.user
if not user.is_authenticated:
return User.objects.none()
elif self.request.method in permissions.SAFE_METHODS or user.is_superuser:
return User.objects.all()
else:
return User.objects.filter(pk=user.id)
def create(self, request, *args, **kwargs):
serializer = UserSerializer(
# FIXME: the method get_serializer_context should be used but it does seem to get lost
context={"request": request}, # the serializer needs the request in the context
data=request.data,
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
@extend_schema_view(
create=extend_schema(summary="Create a new image."),
destroy=extend_schema(summary="Delete an existing image."),
list=extend_schema(summary="List all images."),
partial_update=extend_schema(
summary="Partially update an existing image.",
description="Only `alt_text`, `credits`, and `ppoi` can be updated.",
),
retrieve=extend_schema(summary="Retrieve a single image."),
update=extend_schema(
summary="Update an existing image.",
description="Only `alt_text`, `credits`, and `ppoi` can be updated.",
),
)
class APIImageViewSet(viewsets.ModelViewSet):
serializer_class = ImageSerializer
pagination_class = LimitOffsetPagination
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
def get_queryset(self):
"""The queryset contains all the images if the method is safe, otherwise it only contains
the images owned by the requesting user."""
if self.request.method in permissions.SAFE_METHODS:
return Image.objects.all()
else:
return Image.objects.filter(owner=self.request.user.username)
def create(self, request, *args, **kwargs):
"""Create an Image instance. Any user can create an image."""
serializer = ImageSerializer(
context={"request": request}, # the serializer needs the request in the context
data=request.data,
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
"""Update an Image instance. Only the creator can update an image."""
image = self.get_object()
serializer = ImageSerializer(
image,
context={"request": request}, # the serializer needs the request in the context
data=request.data,
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data)
def destroy(self, request, *args, **kwargs):
"""Destroy an Image instance. Only the owner can delete an image."""
image = self.get_object()
image.delete()
return Response(status=status.HTTP_204_NO_CONTENT)
@extend_schema(
parameters=[
ImageRenderSerializer,
OpenApiParameter(
name="Location",
type=OpenApiTypes.URI,
location=OpenApiParameter.HEADER,
description="/",
response=[301],
),
],
responses={301: None},
)
@decorators.action(["GET"], detail=True)
def render(self, *args, **kwargs):
image = self.get_object()
serializer = ImageRenderSerializer(data=self.request.GET)
if serializer.is_valid():
image_spec = serializer.validated_data
url = image.render(
width=image_spec.get("width", None),
height=image_spec.get("height", None),
)
return HttpResponseRedirect(settings.SITE_URL + url)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
@extend_schema_view(
create=extend_schema(summary="Create a new show."),
retrieve=extend_schema(summary="Retrieve a single show."),
update=extend_schema(summary="Update an existing show."),
partial_update=extend_schema(summary="Partially update an existing show."),
destroy=extend_schema(summary="Delete an existing show."),
list=extend_schema(summary="List all shows."),
)
class APIShowViewSet(viewsets.ModelViewSet):
queryset = Show.objects.all()
serializer_class = ShowSerializer
pagination_class = LimitOffsetPagination
filter_backends = [DjangoFilterBackend, drf_filters.SearchFilter]
filterset_class = filters.ShowFilterSet
search_fields = ["name", "slug", "short_description", "description"]
def get_object(self):
queryset = self.filter_queryset(self.get_queryset())
lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field
lookup_arg = self.kwargs[lookup_url_kwarg]
# allow object retrieval through id or slug
try:
filter_kwargs = {self.lookup_field: int(lookup_arg)}
except ValueError:
filter_kwargs = {"slug": lookup_arg}
obj = get_object_or_404(queryset, **filter_kwargs)
self.check_object_permissions(self.request, obj)
return obj
def create(self, request, *args, **kwargs):
serializer = ShowSerializer(
# FIXME: the method get_serializer_context should be used but it does seem to get lost
context={"request": request}, # the serializer needs the request in the context
data=request.data,
)
if serializer.is_valid(raise_exception=True):
try:
serializer.save()
except IntegrityError:
data = {
"slug": {"code": "unique", "message": "show with this slug already exists."}
}
return Response(data, status=status.HTTP_400_BAD_REQUEST)
return Response(serializer.data, status=status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
partial = kwargs.get("partial", False)
show = self.get_object()
serializer = ShowSerializer(
# FIXME: the method get_serializer_context should be used but it does seem to get lost
context={"request": request}, # the serializer needs the request in the context
data=request.data,
instance=show,
partial=partial,
)
if serializer.is_valid(raise_exception=True):
try:
serializer.save()
return Response(serializer.data)
except IntegrityError:
data = {
"slug": {"code": "unique", "message": "show with this slug already exists."}
}
return Response(data, status=status.HTTP_400_BAD_REQUEST)
def partial_update(self, request, *args, **kwargs):
kwargs["partial"] = True
return self.update(request, *args, **kwargs)
@extend_schema_view(
create=extend_schema(summary="Create a new rrule."),
retrieve=extend_schema(summary="Retrieve a single rrule."),
update=extend_schema(summary="Update an existing rrule."),
partial_update=extend_schema(summary="Partially update an existing rrule."),
destroy=extend_schema(summary="Delete an existing rrule."),
list=extend_schema(summary="List all rrules."),
)
class APIRRuleViewSet(viewsets.ModelViewSet):
queryset = RRule.objects.all()
serializer_class = RRuleSerializer
@extend_schema_view(
create=extend_schema(
summary="Create a new schedule.",
request=ScheduleCreateUpdateRequestSerializer,
responses={
status.HTTP_201_CREATED: OpenApiResponse(
response=ScheduleResponseSerializer,
description=(
"Signals the successful creation of the schedule and of the projected "
"timeslots."
),
),
status.HTTP_202_ACCEPTED: OpenApiResponse(
response=ScheduleDryRunResponseSerializer,
description=(
"Returns the list of timeslots that would be created, updated and deleted if "
"the schedule request would not have been sent with the dryrun flag."
),
),
status.HTTP_400_BAD_REQUEST: OpenApiResponse(
response=ErrorSerializer(many=True),
description=dedent(
"""
Returned in case the request contained invalid data.
This may happen if:
* the last date is before the start date (`no-start-after-end`),
in which case you should correct either the start or until date.
* The start and last date are the same (`no-same-day-start-and-end`).
This is only allowed for single timeslots with the recurrence rule
set to `once`. You should fix either the start or until date.
* The number of conflicts and solutions aren’t the same
(`one-solution-per-conflict`). Only one solution is allowed per conflict,
so you either offered too many or not enough solutions for any reported
conflicts.
* The referenced recurrence rule does not exist.
"""
),
),
status.HTTP_403_FORBIDDEN: OpenApiResponse(
response=ErrorSerializer,
description=(
"Returned in case the request contained no or invalid authenticated data "
"or the authenticated user does not have authorization to perform the "
"requested operation."
),
),
status.HTTP_409_CONFLICT: OpenApiResponse(
response=ScheduleConflictResponseSerializer,
description=dedent(
"""
Returns the list of projected timeslots and any collisions that may have
been found for existing timeslots.
Errors on projected timeslots may include:
* 'This change on the timeslot is not allowed.'
When adding: There was a change in the schedule's data during conflict
resolution.
When updating: Fields 'start', 'end', 'by_weekday' or 'rrule' have changed,
which is not allowed.
* 'No solution given': No solution was provided for the conflict in
`solutions`. Provide a value of `solution_choices`.
* 'Given solution is not accepted for this conflict.':
The solution has a value which is not part of `solution_choices`.
Provide a value of `solution_choices` (at least `ours` or `theirs`).
"""
),
),
},
),
retrieve=extend_schema(summary="Retrieve a single schedule."),
update=extend_schema(
summary="Update an existing schedule.",
request=ScheduleCreateUpdateRequestSerializer,
),
partial_update=extend_schema(
summary="Partially update an existing schedule.",
request=ScheduleCreateUpdateRequestSerializer,
),
destroy=extend_schema(summary="Delete an existing schedule."),
list=extend_schema(summary="List all schedules."),
)
class APIScheduleViewSet(viewsets.ModelViewSet):
filterset_class = filters.ScheduleFilterSet
pagination_class = LimitOffsetPagination
queryset = Schedule.objects.all()
serializer_class = ScheduleSerializer
def get_serializer_class(self):
if self.action in ("create", "update", "partial_update"):
return ScheduleCreateUpdateRequestSerializer
return super().get_serializer_class()
def create(self, request, *args, **kwargs):
"""
Create a schedule, generate timeslots, test for collisions and resolve them
(including notes).
Note that creating or updating a schedule is the only way to create timeslots.
The projected timeslots defined by the schedule are matched against existing
timeslots. The API will return an object that contains
* the schedule's data,
* projected timeslots,
* detected collisions,
* and possible solutions.
As long as no `solutions` object has been set or unresolved collisions exist,
no data is written to the database. A schedule is only created if at least
one timeslot was generated by it.
In order to resolve any possible conflicts, the client must submit a new request with
a solution for each conflict. Possible solutions are listed as part of the projected
timeslot in the `solution_choices` array. In a best-case scenario with no detected
conflicts an empty solutions object will suffice. For more details on the individual
types of solutions see the SolutionChoicesEnum.
**Please note**:
If there's more than one collision for a projected timeslot, only `theirs` and `ours`
are currently supported as solutions.
"""
pk, show_pk = get_values(self.kwargs, "pk", "show_pk")
if show_pk is None:
show_pk = request.data.get("schedule").get("show_id")
# FIXME: this is wrong
if show_pk == 0:
show_pk = request.data.get("show_id")
# Only allow creating when calling /shows/{show_pk}/schedules/ and with ehe `schedule` JSON
# object
if show_pk is None or "schedule" not in request.data:
return Response(status=status.HTTP_400_BAD_REQUEST)
try:
resolution = resolve_conflicts(request.data, pk, show_pk)
# FIXME: Find a better way to do this.
# The exception is thrown by the instantiate_upcoming_schedule function.
except RRule.DoesNotExist as exc:
return JsonResponse({"rruleId": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
except ScheduleConflictError as exc:
return Response(exc.conflicts, status.HTTP_409_CONFLICT)
if all(key in resolution for key in ["create", "update", "delete"]):
# this is a dry-run
return Response(resolution, status=status.HTTP_202_ACCEPTED)
return Response(resolution, status=status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
"""
Update a schedule, generate timeslots, test for collisions and resolve
them including notes.
"""
if not request.user.is_superuser:
return Response(status=status.HTTP_401_UNAUTHORIZED)
if request.method == "PATCH":
# only these fields can be updated without generating conflicts
allowed = {"default_playlist_id", "is_repetition", "last_date"}
if set(request.data.keys()).issubset(allowed):
schedule = self.get_object()
if default_playlist_id := request.data.get("default_playlist_id"):
if default_playlist_id == "":
# "clear" the default_playlist_id if the field has no value
schedule.default_playlist_id = None
else:
schedule.default_playlist_id = int(default_playlist_id)
if is_repetition := request.data.get("is_repetition"):
if is_repetition == "true" or is_repetition == "1":
schedule.is_repetition = True
if is_repetition == "false" or is_repetition == "0":
schedule.is_repetition = False
if last_date := request.data.get("last_date"):
last_date = date.fromisoformat(last_date)
if schedule.last_date is None or schedule.last_date > last_date:
schedule.last_date = last_date
last_end = timezone.make_aware(
datetime.combine(last_date, schedule.end_time)
)
TimeSlot.objects.filter(schedule=schedule, start__gt=last_end).delete()
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
schedule.save()
serializer = ScheduleSerializer(schedule)
return Response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
# Only allow updating when with the `schedule` JSON object
if "schedule" not in request.data:
return Response(status=status.HTTP_400_BAD_REQUEST)
schedule = self.get_object()
# FIXME: this is redundant now and should be removed
# If default playlist id or repetition are given, just update
if default_playlist_id := request.data.get("schedule").get("default_playlist_id"):
schedule.default_playlist_id = int(default_playlist_id)
schedule.save()
serializer = ScheduleSerializer(schedule)
return Response(serializer.data)
if is_repetition := request.data.get("schedule").get("is_repetition"):
schedule.is_repetition = bool(is_repetition)
schedule.save()
serializer = ScheduleSerializer(schedule)
return Response(serializer.data)
try:
resolution = resolve_conflicts(request.data, schedule.pk, schedule.show.pk)
except ScheduleConflictError as exc:
return Response(exc.conflicts, status.HTTP_409_CONFLICT)
return Response(resolution)
# TODO: Create is currently not implemented because timeslots are supposed to be inserted
# by creating or updating a schedule.
# There might be a use case for adding a single timeslot without any conflicts though.
@extend_schema_view(
retrieve=extend_schema(summary="Retrieve a single timeslot."),
update=extend_schema(summary="Update an existing timeslot."),
partial_update=extend_schema(summary="Partially update an existing timeslot."),
destroy=extend_schema(summary="Delete an existing timeslot."),
list=extend_schema(
summary="List all timeslots.",
description=dedent(
"""
By default, only timeslots ranging from now + 60 days will be displayed.
You may override this default overriding start and/or end parameter.
"""
),
),
)
class APITimeSlotViewSet(
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
viewsets.GenericViewSet,
):
filterset_class = filters.TimeSlotFilterSet
pagination_class = LimitOffsetPagination
queryset = TimeSlot.objects.all().order_by("-start")
serializer_class = TimeSlotSerializer
def update(self, request, *args, **kwargs):
show_pk = get_values(self.kwargs, "show_pk")
timeslot = self.get_object()
serializer = TimeSlotSerializer(timeslot, data=request.data)
if serializer.is_valid(raise_exception=True):
serializer.save()
# Return the next repetition
# We do this because the Dashboard needs to update the repetition timeslot as well
# but with another playlist containing the recording instead of the original playlist
if first_repetition := TimeSlot.objects.filter(
schedule__show=show_pk, start__gt=timeslot.start, repetition_of=timeslot
).first():
serializer = TimeSlotSerializer(first_repetition)
return Response(serializer.data)
# ...or nothing if there isn't one
return Response(serializer.data, status=status.HTTP_200_OK)
@extend_schema_view(
create=extend_schema(summary="Create a new note."),
retrieve=extend_schema(summary="Retrieve a single note."),
update=extend_schema(summary="Update an existing note."),
partial_update=extend_schema(
summary="Partially update an existing note.",
description="Only admins can partially update existing notes.",
),
destroy=extend_schema(summary="Delete an existing note."),
list=extend_schema(summary="List all notes."),
)
class APINoteViewSet(viewsets.ModelViewSet):
filterset_class = filters.NoteFilterSet
pagination_class = LimitOffsetPagination
serializer_class = NoteSerializer
def get_queryset(self):
"""The queryset contains all the notes if the method is safe or the requesting user is a
superuser, otherwise it only contains the notes for shows owned by the requesting user."""
user = self.request.user
if self.request.method in permissions.SAFE_METHODS or user.has_perm("program.update_note"):
return Note.objects.all()
else:
return Note.objects.filter(timeslot__schedule__show__owners=user)
class ActiveFilterMixin:
filter_class = filters.ActiveFilterSet
@extend_schema_view(
create=extend_schema(summary="Create a new category."),
retrieve=extend_schema(summary="Retrieve a single category."),
update=extend_schema(summary="Update an existing category."),
partial_update=extend_schema(summary="Partially update an existing category."),
destroy=extend_schema(summary="Delete an existing category."),
list=extend_schema(summary="List all categories."),
)
class APICategoryViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = Category.objects.all()
serializer_class = CategorySerializer
@extend_schema_view(
create=extend_schema(summary="Create a new type."),
retrieve=extend_schema(summary="Retrieve a single type."),
update=extend_schema(summary="Update an existing type."),
partial_update=extend_schema(summary="Partially update an existing type."),
destroy=extend_schema(summary="Delete an existing type."),
list=extend_schema(summary="List all types."),
)
class APITypeViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = Type.objects.all()
serializer_class = TypeSerializer
@extend_schema_view(
create=extend_schema(summary="Create a new topic."),
retrieve=extend_schema(summary="Retrieve a single topic."),
update=extend_schema(summary="Update an existing topic."),
partial_update=extend_schema(summary="Partially update an existing topic."),
destroy=extend_schema(summary="Delete an existing topic."),
list=extend_schema(summary="List all topics."),
)
class APITopicViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = Topic.objects.all()
serializer_class = TopicSerializer
@extend_schema_view(
create=extend_schema(summary="Create a new music focus."),
retrieve=extend_schema(summary="Retrieve a single music focus."),
update=extend_schema(summary="Update an existing music focus."),
partial_update=extend_schema(summary="Partially update an existing music focus."),
destroy=extend_schema(summary="Delete an existing music focus."),
list=extend_schema(summary="List all music focuses."),
)
class APIMusicFocusViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = MusicFocus.objects.all()
serializer_class = MusicFocusSerializer
@extend_schema_view(
create=extend_schema(summary="Create a new funding category."),
retrieve=extend_schema(summary="Retrieve a single funding category."),
update=extend_schema(summary="Update an existing funding category."),
partial_update=extend_schema(summary="Partially update an existing funding category."),
destroy=extend_schema(summary="Delete an existing funding category."),
list=extend_schema(summary="List all funding categories."),
)
class APIFundingCategoryViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = FundingCategory.objects.all()
serializer_class = FundingCategorySerializer
@extend_schema_view(
create=extend_schema(summary="Create a new language."),
retrieve=extend_schema(summary="Retrieve a single language."),
update=extend_schema(summary="Update an existing language."),
partial_update=extend_schema(summary="Partially update an existing language."),
destroy=extend_schema(summary="Delete an existing language."),
list=extend_schema(summary="List all languages."),
)
class APILanguageViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = Language.objects.all()
serializer_class = LanguageSerializer
@extend_schema_view(
create=extend_schema(summary="Create a new host."),
retrieve=extend_schema(summary="Retrieve a single host."),
update=extend_schema(summary="Update an existing host."),
partial_update=extend_schema(summary="Partially update an existing host."),
destroy=extend_schema(summary="Delete an existing host."),
list=extend_schema(summary="List all hosts."),
)
class APIHostViewSet(ActiveFilterMixin, viewsets.ModelViewSet):
queryset = Host.objects.all().order_by("-is_active", "name")
serializer_class = HostSerializer
pagination_class = LimitOffsetPagination
filter_backends = [drf_filters.SearchFilter]
search_fields = ["name", "email"]
def create(self, request, *args, **kwargs):
serializer = HostSerializer(
# FIXME: the method get_serializer_context should be used but it does seem to get lost
context={"request": request}, # the serializer needs the request in the context
data=request.data,
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
def update(self, request, *args, **kwargs):
partial = kwargs.get("partial", False)
host = self.get_object()
serializer = HostSerializer(
# FIXME: the method get_serializer_context should be used but it does seem to get lost
context={"request": request}, # the serializer needs the request in the context
data=request.data,
instance=host,
partial=partial,
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data, status=status.HTTP_200_OK)
def partial_update(self, request, *args, **kwargs):
kwargs["partial"] = True
return self.update(request, *args, **kwargs)
@extend_schema_view(
create=extend_schema(summary="Create a new link type."),
retrieve=extend_schema(summary="Retrieve a single link type."),
update=extend_schema(summary="Update an existing link type."),
partial_update=extend_schema(summary="Partially update an existing link type."),
destroy=extend_schema(summary="Delete an existing link type."),
list=extend_schema(summary="List all link types."),
)
class APILinkTypeViewSet(viewsets.ModelViewSet):
queryset = LinkType.objects.all()
serializer_class = LinkTypeSerializer
def destroy(self, request, *args, **kwargs):
"""Destroying a link type just makes is inactive.
This is needed to preserve all the links that reference a link type."""
link_type = self.get_object()
link_type.is_active = False
link_type.save()
return Response(status=status.HTTP_204_NO_CONTENT)
@extend_schema_view(
create=extend_schema(summary="Create a new license type."),
retrieve=extend_schema(summary="Retrieve a single license type."),
update=extend_schema(summary="Update an existing license type."),
partial_update=extend_schema(summary="Partially update an existing license type."),
destroy=extend_schema(summary="Delete an existing license type."),
list=extend_schema(summary="List all license types."),
)
class APILicenseViewSet(viewsets.ModelViewSet):
queryset = License.objects.all()
serializer_class = LicenseSerializer
@extend_schema_view(
list=extend_schema(summary="List all settings."),
)
class APIRadioSettingsViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
queryset = RadioSettings.objects.all()
serializer_class = RadioSettingsSerializer