Module fl_server_api.views.group

View Source

# SPDX-FileCopyrightText: 2024 Benedikt Franke <benedikt.franke@dlr.de>
# SPDX-FileCopyrightText: 2024 Florian Heinrich <florian.heinrich@dlr.de>
#
# SPDX-License-Identifier: Apache-2.0

from django.contrib.auth.models import Group as GroupModel
from django.http import HttpRequest, HttpResponse
from drf_spectacular.utils import extend_schema, OpenApiExample, OpenApiParameter
from rest_framework import status
from rest_framework.exceptions import PermissionDenied
from rest_framework.response import Response

from fl_server_core.models import User as UserModel

from .base import ViewSet
from ..utils import get_entity
from ..serializers.generic import ErrorSerializer, GroupSerializer
from ..openapi import error_response_403


_default_group_example = OpenApiExample(
    name="Get group by id",
    description="\n\n".join([
        "Retrieve group data by group ID.",
        "_Please not that the user Jane Doe has to be created and authorized first._",
    ]),
    value=1,
    parameter_only=("id", OpenApiParameter.PATH)
)
"""Default OpenAPI Specification example for getting a group by its ID."""


class Group(ViewSet):
    """
    Group Model ViewSet.
    """

    serializer_class = GroupSerializer
    """The serializer for the ViewSet."""

    def _get_group(self, user: UserModel, group_id: int) -> GroupModel:
        """
        Get a group by id if the user is a member of the group.

        Args:
            user (UserModel): The user making the request.
            group_id (int): The id of the group.

        Raises:
            PermissionDenied: If the user is not a member of the group.

        Returns:
            GroupModel: The group instance.
        """
        group = get_entity(GroupModel, pk=group_id)
        if not user.groups.contains(group):
            raise PermissionDenied("You are not allowed to access this group.")
        return group

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        }
    )
    def list(self, request: HttpRequest) -> HttpResponse:
        """
        Get all groups.

        Args:
            request (HttpRequest): request object

        Raises:
            PermissionDenied: If user is not a superuser.

        Returns:
            HttpResponse: list of groups as json response
        """
        if not request.user.is_superuser:
            raise PermissionDenied("You are not allowed to access all groups.")
        groups = GroupModel.objects.all()
        serializer = GroupSerializer(groups, many=True)
        return Response(serializer.data)

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[_default_group_example]
    )
    def retrieve(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Get group information by id.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: group as json response
        """
        group = self._get_group(request.user, id)
        serializer = GroupSerializer(group)
        return Response(serializer.data)

    @extend_schema(
        responses={
            status.HTTP_201_CREATED: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[OpenApiExample(
            name="Create group",
            description="Create a new group.",
            value={"name": "My new amazing group"},
        )]
    )
    def create(self, request: HttpRequest) -> HttpResponse:
        """
        Create a new group.

        Args:
            request (HttpRequest): request object

        Returns:
            HttpResponse: new created group as json response
        """
        serializer = GroupSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def _update(self, request: HttpRequest, id: int, *, partial: bool) -> HttpResponse:
        """
        Update group information.

        Args:
            request (HttpRequest): request object
            id (int): group id
            partial (bool): allow partial update

        Returns:
            HttpResponse: updated group as json response
        """
        group = self._get_group(request.user, id)
        serializer = GroupSerializer(group, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        if getattr(group, '_prefetched_objects_cache', None):
            # If 'prefetch_related' has been applied to a queryset, we need to
            # forcibly invalidate the prefetch cache on the instance.
            group._prefetched_objects_cache = {}
        return Response(serializer.data)

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[
            _default_group_example,
            OpenApiExample(
                name="Update group",
                description="Update group fields.",
                value={"name": "My new amazing group is the best!"},
            )
        ]
    )
    def update(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Update group information.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: updated group as json response
        """
        return self._update(request, id, partial=False)

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[
            _default_group_example,
            OpenApiExample(
                name="Update group partially",
                description="Update only some group fields.",
                value={"name": "My new amazing group is the best!"},
            )
        ]
    )
    def partial_update(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Update group information partially.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: updated group as json response
        """
        return self._update(request, id, partial=True)

    @extend_schema(
        responses={
            status.HTTP_204_NO_CONTENT: None,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[_default_group_example]
    )
    def destroy(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Remove group by id.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: 204 NO CONTENT
        """
        group = self._get_group(request.user, id)
        group.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

Classes

Group

class Group(
    **kwargs
)

Group Model ViewSet.

View Source

class Group(ViewSet):
    """
    Group Model ViewSet.
    """

    serializer_class = GroupSerializer
    """The serializer for the ViewSet."""

    def _get_group(self, user: UserModel, group_id: int) -> GroupModel:
        """
        Get a group by id if the user is a member of the group.

        Args:
            user (UserModel): The user making the request.
            group_id (int): The id of the group.

        Raises:
            PermissionDenied: If the user is not a member of the group.

        Returns:
            GroupModel: The group instance.
        """
        group = get_entity(GroupModel, pk=group_id)
        if not user.groups.contains(group):
            raise PermissionDenied("You are not allowed to access this group.")
        return group

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        }
    )
    def list(self, request: HttpRequest) -> HttpResponse:
        """
        Get all groups.

        Args:
            request (HttpRequest): request object

        Raises:
            PermissionDenied: If user is not a superuser.

        Returns:
            HttpResponse: list of groups as json response
        """
        if not request.user.is_superuser:
            raise PermissionDenied("You are not allowed to access all groups.")
        groups = GroupModel.objects.all()
        serializer = GroupSerializer(groups, many=True)
        return Response(serializer.data)

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[_default_group_example]
    )
    def retrieve(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Get group information by id.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: group as json response
        """
        group = self._get_group(request.user, id)
        serializer = GroupSerializer(group)
        return Response(serializer.data)

    @extend_schema(
        responses={
            status.HTTP_201_CREATED: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[OpenApiExample(
            name="Create group",
            description="Create a new group.",
            value={"name": "My new amazing group"},
        )]
    )
    def create(self, request: HttpRequest) -> HttpResponse:
        """
        Create a new group.

        Args:
            request (HttpRequest): request object

        Returns:
            HttpResponse: new created group as json response
        """
        serializer = GroupSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def _update(self, request: HttpRequest, id: int, *, partial: bool) -> HttpResponse:
        """
        Update group information.

        Args:
            request (HttpRequest): request object
            id (int): group id
            partial (bool): allow partial update

        Returns:
            HttpResponse: updated group as json response
        """
        group = self._get_group(request.user, id)
        serializer = GroupSerializer(group, data=request.data, partial=partial)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        if getattr(group, '_prefetched_objects_cache', None):
            # If 'prefetch_related' has been applied to a queryset, we need to
            # forcibly invalidate the prefetch cache on the instance.
            group._prefetched_objects_cache = {}
        return Response(serializer.data)

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[
            _default_group_example,
            OpenApiExample(
                name="Update group",
                description="Update group fields.",
                value={"name": "My new amazing group is the best!"},
            )
        ]
    )
    def update(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Update group information.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: updated group as json response
        """
        return self._update(request, id, partial=False)

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[
            _default_group_example,
            OpenApiExample(
                name="Update group partially",
                description="Update only some group fields.",
                value={"name": "My new amazing group is the best!"},
            )
        ]
    )
    def partial_update(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Update group information partially.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: updated group as json response
        """
        return self._update(request, id, partial=True)

    @extend_schema(
        responses={
            status.HTTP_204_NO_CONTENT: None,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[_default_group_example]
    )
    def destroy(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Remove group by id.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: 204 NO CONTENT
        """
        group = self._get_group(request.user, id)
        group.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

---

Ancestors (in MRO)

• fl_server_api.views.base.ViewSet
• rest_framework.viewsets.ViewSet
• rest_framework.viewsets.ViewSetMixin
• rest_framework.views.APIView
• django.views.generic.base.View

Class variables

authentication_classes

basename

content_negotiation_class

description

detail

http_method_names

metadata_class

name

parser_classes

permission_classes

renderer_classes

schema

serializer_class

The serializer for the ViewSet.

settings

suffix

throttle_classes

versioning_class

Static methods

as_view

def as_view(
    actions=None,
    **initkwargs
)

Because of the way class based views create a closure around the

instantiated view, we need to totally reimplement .as_view, and slightly modify the view function that is created and returned.

View Source

    @classonlymethod
    def as_view(cls, actions=None, **initkwargs):
        """
        Because of the way class based views create a closure around the
        instantiated view, we need to totally reimplement `.as_view`,
        and slightly modify the view function that is created and returned.
        """
        # The name and description initkwargs may be explicitly overridden for
        # certain route configurations. eg, names of extra actions.
        cls.name = None
        cls.description = None

        # The suffix initkwarg is reserved for displaying the viewset type.
        # This initkwarg should have no effect if the name is provided.
        # eg. 'List' or 'Instance'.
        cls.suffix = None

        # The detail initkwarg is reserved for introspecting the viewset type.
        cls.detail = None

        # Setting a basename allows a view to reverse its action urls. This
        # value is provided by the router through the initkwargs.
        cls.basename = None

        # actions must not be empty
        if not actions:
            raise TypeError("The `actions` argument must be provided when "
                            "calling `.as_view()` on a ViewSet. For example "
                            "`.as_view({'get': 'list'})`")

        # sanitize keyword arguments
        for key in initkwargs:
            if key in cls.http_method_names:
                raise TypeError("You tried to pass in the %s method name as a "
                                "keyword argument to %s(). Don't do that."
                                % (key, cls.__name__))
            if not hasattr(cls, key):
                raise TypeError("%s() received an invalid keyword %r" % (
                    cls.__name__, key))

        # name and suffix are mutually exclusive
        if 'name' in initkwargs and 'suffix' in initkwargs:
            raise TypeError("%s() received both `name` and `suffix`, which are "
                            "mutually exclusive arguments." % (cls.__name__))

        def view(request, *args, **kwargs):
            self = cls(**initkwargs)

            if 'get' in actions and 'head' not in actions:
                actions['head'] = actions['get']

            # We also store the mapping of request methods to actions,
            # so that we can later set the action attribute.
            # eg. `self.action = 'list'` on an incoming GET request.
            self.action_map = actions

            # Bind methods to actions
            # This is the bit that's different to a standard view
            for method, action in actions.items():
                handler = getattr(self, action)
                setattr(self, method, handler)

            self.request = request
            self.args = args
            self.kwargs = kwargs

            # And continue as usual
            return self.dispatch(request, *args, **kwargs)

        # take name and docstring from class
        update_wrapper(view, cls, updated=())

        # and possible attributes set by decorators
        # like csrf_exempt from dispatch
        update_wrapper(view, cls.dispatch, assigned=())

        # We need to set these on the view function, so that breadcrumb
        # generation can pick out these bits of information from a
        # resolved URL.
        view.cls = cls
        view.initkwargs = initkwargs
        view.actions = actions
        return csrf_exempt(view)

get_extra_actions

def get_extra_actions()

Get the methods that are marked as an extra ViewSet @action.

View Source

    @classmethod
    def get_extra_actions(cls):
        """
        Get the methods that are marked as an extra ViewSet `@action`.
        """
        return [_check_attr_name(method, name)
                for name, method
                in getmembers(cls, _is_extra_action)]

Instance variables

allowed_methods

Wrap Django's private _allowed_methods interface in a public property.

default_response_headers

Methods

check_object_permissions

def check_object_permissions(
    self,
    request,
    obj
)

Check if the request should be permitted for a given object.

Raises an appropriate exception if the request is not permitted.

View Source

    def check_object_permissions(self, request, obj):
        """
        Check if the request should be permitted for a given object.
        Raises an appropriate exception if the request is not permitted.
        """
        for permission in self.get_permissions():
            if not permission.has_object_permission(request, self, obj):
                self.permission_denied(
                    request,
                    message=getattr(permission, 'message', None),
                    code=getattr(permission, 'code', None)
                )

check_permissions

def check_permissions(
    self,
    request
)

Check if the request should be permitted.

Raises an appropriate exception if the request is not permitted.

View Source

    def check_permissions(self, request):
        """
        Check if the request should be permitted.
        Raises an appropriate exception if the request is not permitted.
        """
        for permission in self.get_permissions():
            if not permission.has_permission(request, self):
                self.permission_denied(
                    request,
                    message=getattr(permission, 'message', None),
                    code=getattr(permission, 'code', None)
                )

check_throttles

def check_throttles(
    self,
    request
)

Check if request should be throttled.

Raises an appropriate exception if the request is throttled.

View Source

    def check_throttles(self, request):
        """
        Check if request should be throttled.
        Raises an appropriate exception if the request is throttled.
        """
        throttle_durations = []
        for throttle in self.get_throttles():
            if not throttle.allow_request(request, self):
                throttle_durations.append(throttle.wait())

        if throttle_durations:
            # Filter out `None` values which may happen in case of config / rate
            # changes, see #1438
            durations = [
                duration for duration in throttle_durations
                if duration is not None
            ]

            duration = max(durations, default=None)
            self.throttled(request, duration)

create

def create(
    self,
    request: django.http.request.HttpRequest
) -> django.http.response.HttpResponse

Create a new group.

Parameters:

Name	Type	Description	Default
request	HttpRequest	request object	None

Returns:

Type	Description
HttpResponse	new created group as json response

View Source

    @extend_schema(
        responses={
            status.HTTP_201_CREATED: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[OpenApiExample(
            name="Create group",
            description="Create a new group.",
            value={"name": "My new amazing group"},
        )]
    )
    def create(self, request: HttpRequest) -> HttpResponse:
        """
        Create a new group.

        Args:
            request (HttpRequest): request object

        Returns:
            HttpResponse: new created group as json response
        """
        serializer = GroupSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data, status=status.HTTP_201_CREATED)

destroy

def destroy(
    self,
    request: django.http.request.HttpRequest,
    id: int
) -> django.http.response.HttpResponse

Remove group by id.

Parameters:

Name	Type	Description	Default
request	HttpRequest	request object	None
id	int	group id	None

Returns:

Type	Description
HttpResponse	204 NO CONTENT

View Source

    @extend_schema(
        responses={
            status.HTTP_204_NO_CONTENT: None,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[_default_group_example]
    )
    def destroy(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Remove group by id.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: 204 NO CONTENT
        """
        group = self._get_group(request.user, id)
        group.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

determine_version

def determine_version(
    self,
    request,
    *args,
    **kwargs
)

If versioning is being used, then determine any API version for the

incoming request. Returns a two-tuple of (version, versioning_scheme)

View Source

    def determine_version(self, request, *args, **kwargs):
        """
        If versioning is being used, then determine any API version for the
        incoming request. Returns a two-tuple of (version, versioning_scheme)
        """
        if self.versioning_class is None:
            return (None, None)
        scheme = self.versioning_class()
        return (scheme.determine_version(request, *args, **kwargs), scheme)

dispatch

def dispatch(
    self,
    request,
    *args,
    **kwargs
)

.dispatch() is pretty much the same as Django's regular dispatch,

but with extra hooks for startup, finalize, and exception handling.

View Source

    def dispatch(self, request, *args, **kwargs):
        """
        `.dispatch()` is pretty much the same as Django's regular dispatch,
        but with extra hooks for startup, finalize, and exception handling.
        """
        self.args = args
        self.kwargs = kwargs
        request = self.initialize_request(request, *args, **kwargs)
        self.request = request
        self.headers = self.default_response_headers  # deprecate?

        try:
            self.initial(request, *args, **kwargs)

            # Get the appropriate handler method
            if request.method.lower() in self.http_method_names:
                handler = getattr(self, request.method.lower(),
                                  self.http_method_not_allowed)
            else:
                handler = self.http_method_not_allowed

            response = handler(request, *args, **kwargs)

        except Exception as exc:
            response = self.handle_exception(exc)

        self.response = self.finalize_response(request, response, *args, **kwargs)
        return self.response

finalize_response

def finalize_response(
    self,
    request,
    response,
    *args,
    **kwargs
)

Returns the final response object.

View Source

    def finalize_response(self, request, response, *args, **kwargs):
        """
        Returns the final response object.
        """
        # Make the error obvious if a proper response is not returned
        assert isinstance(response, HttpResponseBase), (
            'Expected a `Response`, `HttpResponse` or `HttpStreamingResponse` '
            'to be returned from the view, but received a `%s`'
            % type(response)
        )

        if isinstance(response, Response):
            if not getattr(request, 'accepted_renderer', None):
                neg = self.perform_content_negotiation(request, force=True)
                request.accepted_renderer, request.accepted_media_type = neg

            response.accepted_renderer = request.accepted_renderer
            response.accepted_media_type = request.accepted_media_type
            response.renderer_context = self.get_renderer_context()

        # Add new vary headers to the response instead of overwriting.
        vary_headers = self.headers.pop('Vary', None)
        if vary_headers is not None:
            patch_vary_headers(response, cc_delim_re.split(vary_headers))

        for key, value in self.headers.items():
            response[key] = value

        return response

get_authenticate_header

def get_authenticate_header(
    self,
    request
)

If a request is unauthenticated, determine the WWW-Authenticate

header to use for 401 responses, if any.

View Source

    def get_authenticate_header(self, request):
        """
        If a request is unauthenticated, determine the WWW-Authenticate
        header to use for 401 responses, if any.
        """
        authenticators = self.get_authenticators()
        if authenticators:
            return authenticators[0].authenticate_header(request)

get_authenticators

def get_authenticators(
    self
)

Get the authenticators for the ViewSet.

This method gets the view method and, if it has authentication classes defined via the decorator, returns them. Otherwise, it falls back to the default authenticators.

Returns:

Type	Description
list	The authenticators for the ViewSet.

View Source

    def get_authenticators(self):
        """
        Get the authenticators for the ViewSet.

        This method gets the view method and, if it has authentication classes defined via the decorator, returns them.
        Otherwise, it falls back to the default authenticators.

        Returns:
            list: The authenticators for the ViewSet.
        """
        if method := self._get_view_method():
            if hasattr(method, "authentication_classes"):
                return method.authentication_classes
        return super().get_authenticators()

get_content_negotiator

def get_content_negotiator(
    self
)

Instantiate and return the content negotiation class to use.

View Source

    def get_content_negotiator(self):
        """
        Instantiate and return the content negotiation class to use.
        """
        if not getattr(self, '_negotiator', None):
            self._negotiator = self.content_negotiation_class()
        return self._negotiator

get_exception_handler

def get_exception_handler(
    self
)

Returns the exception handler that this view uses.

View Source

    def get_exception_handler(self):
        """
        Returns the exception handler that this view uses.
        """
        return self.settings.EXCEPTION_HANDLER

get_exception_handler_context

def get_exception_handler_context(
    self
)

Returns a dict that is passed through to EXCEPTION_HANDLER,

as the context argument.

View Source

    def get_exception_handler_context(self):
        """
        Returns a dict that is passed through to EXCEPTION_HANDLER,
        as the `context` argument.
        """
        return {
            'view': self,
            'args': getattr(self, 'args', ()),
            'kwargs': getattr(self, 'kwargs', {}),
            'request': getattr(self, 'request', None)
        }

get_extra_action_url_map

def get_extra_action_url_map(
    self
)

Build a map of {names: urls} for the extra actions.

This method will noop if detail was not provided as a view initkwarg.

View Source

    def get_extra_action_url_map(self):
        """
        Build a map of {names: urls} for the extra actions.

        This method will noop if `detail` was not provided as a view initkwarg.
        """
        action_urls = OrderedDict()

        # exit early if `detail` has not been provided
        if self.detail is None:
            return action_urls

        # filter for the relevant extra actions
        actions = [
            action for action in self.get_extra_actions()
            if action.detail == self.detail
        ]

        for action in actions:
            try:
                url_name = '%s-%s' % (self.basename, action.url_name)
                namespace = self.request.resolver_match.namespace
                if namespace:
                    url_name = '%s:%s' % (namespace, url_name)

                url = reverse(url_name, self.args, self.kwargs, request=self.request)
                view = self.__class__(**action.kwargs)
                action_urls[view.get_view_name()] = url
            except NoReverseMatch:
                pass  # URL requires additional arguments, ignore

        return action_urls

get_format_suffix

def get_format_suffix(
    self,
    **kwargs
)

Determine if the request includes a '.json' style format suffix

View Source

    def get_format_suffix(self, **kwargs):
        """
        Determine if the request includes a '.json' style format suffix
        """
        if self.settings.FORMAT_SUFFIX_KWARG:
            return kwargs.get(self.settings.FORMAT_SUFFIX_KWARG)

get_parser_context

def get_parser_context(
    self,
    http_request
)

Returns a dict that is passed through to Parser.parse(),

as the parser_context keyword argument.

View Source

    def get_parser_context(self, http_request):
        """
        Returns a dict that is passed through to Parser.parse(),
        as the `parser_context` keyword argument.
        """
        # Note: Additionally `request` and `encoding` will also be added
        #       to the context by the Request object.
        return {
            'view': self,
            'args': getattr(self, 'args', ()),
            'kwargs': getattr(self, 'kwargs', {})
        }

get_parsers

def get_parsers(
    self
)

Instantiates and returns the list of parsers that this view can use.

View Source

    def get_parsers(self):
        """
        Instantiates and returns the list of parsers that this view can use.
        """
        return [parser() for parser in self.parser_classes]

get_permissions

def get_permissions(
    self
)

Get the permissions for the ViewSet.

This method gets the view method and, if it has permission classes defined via the decorator, returns them. Otherwise, it falls back to the default permissions.

Returns:

Type	Description
list	The permissions for the ViewSet.

View Source

    def get_permissions(self):
        """
        Get the permissions for the ViewSet.

        This method gets the view method and, if it has permission classes defined via the decorator, returns them.
        Otherwise, it falls back to the default permissions.

        Returns:
            list: The permissions for the ViewSet.
        """
        if method := self._get_view_method():
            if hasattr(method, "permission_classes"):
                return method.permission_classes
        return super().get_permissions()

get_renderer_context

def get_renderer_context(
    self
)

Returns a dict that is passed through to Renderer.render(),

as the renderer_context keyword argument.

View Source

    def get_renderer_context(self):
        """
        Returns a dict that is passed through to Renderer.render(),
        as the `renderer_context` keyword argument.
        """
        # Note: Additionally 'response' will also be added to the context,
        #       by the Response object.
        return {
            'view': self,
            'args': getattr(self, 'args', ()),
            'kwargs': getattr(self, 'kwargs', {}),
            'request': getattr(self, 'request', None)
        }

get_renderers

def get_renderers(
    self
)

Instantiates and returns the list of renderers that this view can use.

View Source

    def get_renderers(self):
        """
        Instantiates and returns the list of renderers that this view can use.
        """
        return [renderer() for renderer in self.renderer_classes]

get_throttles

def get_throttles(
    self
)

Instantiates and returns the list of throttles that this view uses.

View Source

    def get_throttles(self):
        """
        Instantiates and returns the list of throttles that this view uses.
        """
        return [throttle() for throttle in self.throttle_classes]

get_view_description

def get_view_description(
    self,
    html=False
)

Return some descriptive text for the view, as used in OPTIONS responses

and in the browsable API.

View Source

    def get_view_description(self, html=False):
        """
        Return some descriptive text for the view, as used in OPTIONS responses
        and in the browsable API.
        """
        func = self.settings.VIEW_DESCRIPTION_FUNCTION
        return func(self, html)

get_view_name

def get_view_name(
    self
)

Return the view name, as used in OPTIONS responses and in the

browsable API.

View Source

    def get_view_name(self):
        """
        Return the view name, as used in OPTIONS responses and in the
        browsable API.
        """
        func = self.settings.VIEW_NAME_FUNCTION
        return func(self)

handle_exception

def handle_exception(
    self,
    exc
)

Handle any exception that occurs, by returning an appropriate response,

or re-raising the error.

View Source

    def handle_exception(self, exc):
        """
        Handle any exception that occurs, by returning an appropriate response,
        or re-raising the error.
        """
        if isinstance(exc, (exceptions.NotAuthenticated,
                            exceptions.AuthenticationFailed)):
            # WWW-Authenticate header for 401 responses, else coerce to 403
            auth_header = self.get_authenticate_header(self.request)

            if auth_header:
                exc.auth_header = auth_header
            else:
                exc.status_code = status.HTTP_403_FORBIDDEN

        exception_handler = self.get_exception_handler()

        context = self.get_exception_handler_context()
        response = exception_handler(exc, context)

        if response is None:
            self.raise_uncaught_exception(exc)

        response.exception = True
        return response

http_method_not_allowed

def http_method_not_allowed(
    self,
    request,
    *args,
    **kwargs
)

If request.method does not correspond to a handler method,

determine what kind of exception to raise.

View Source

    def http_method_not_allowed(self, request, *args, **kwargs):
        """
        If `request.method` does not correspond to a handler method,
        determine what kind of exception to raise.
        """
        raise exceptions.MethodNotAllowed(request.method)

initial

def initial(
    self,
    request,
    *args,
    **kwargs
)

Runs anything that needs to occur prior to calling the method handler.

View Source

    def initial(self, request, *args, **kwargs):
        """
        Runs anything that needs to occur prior to calling the method handler.
        """
        self.format_kwarg = self.get_format_suffix(**kwargs)

        # Perform content negotiation and store the accepted info on the request
        neg = self.perform_content_negotiation(request)
        request.accepted_renderer, request.accepted_media_type = neg

        # Determine the API version, if versioning is in use.
        version, scheme = self.determine_version(request, *args, **kwargs)
        request.version, request.versioning_scheme = version, scheme

        # Ensure that the incoming request is permitted
        self.perform_authentication(request)
        self.check_permissions(request)
        self.check_throttles(request)

initialize_request

def initialize_request(
    self,
    request,
    *args,
    **kwargs
)

Set the .action attribute on the view, depending on the request method.

View Source

    def initialize_request(self, request, *args, **kwargs):
        """
        Set the `.action` attribute on the view, depending on the request method.
        """
        request = super().initialize_request(request, *args, **kwargs)
        method = request.method.lower()
        if method == 'options':
            # This is a special case as we always provide handling for the
            # options method in the base `View` class.
            # Unlike the other explicitly defined actions, 'metadata' is implicit.
            self.action = 'metadata'
        else:
            self.action = self.action_map.get(method)
        return request

list

def list(
    self,
    request: django.http.request.HttpRequest
) -> django.http.response.HttpResponse

Get all groups.

Parameters:

Name	Type	Description	Default
request	HttpRequest	request object	None

Returns:

Type	Description
HttpResponse	list of groups as json response

Raises:

Type	Description
PermissionDenied	If user is not a superuser.

View Source

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        }
    )
    def list(self, request: HttpRequest) -> HttpResponse:
        """
        Get all groups.

        Args:
            request (HttpRequest): request object

        Raises:
            PermissionDenied: If user is not a superuser.

        Returns:
            HttpResponse: list of groups as json response
        """
        if not request.user.is_superuser:
            raise PermissionDenied("You are not allowed to access all groups.")
        groups = GroupModel.objects.all()
        serializer = GroupSerializer(groups, many=True)
        return Response(serializer.data)

options

def options(
    self,
    request,
    *args,
    **kwargs
)

Handler method for HTTP 'OPTIONS' request.

View Source

    def options(self, request, *args, **kwargs):
        """
        Handler method for HTTP 'OPTIONS' request.
        """
        if self.metadata_class is None:
            return self.http_method_not_allowed(request, *args, **kwargs)
        data = self.metadata_class().determine_metadata(request, self)
        return Response(data, status=status.HTTP_200_OK)

partial_update

def partial_update(
    self,
    request: django.http.request.HttpRequest,
    id: int
) -> django.http.response.HttpResponse

Update group information partially.

Parameters:

Name	Type	Description	Default
request	HttpRequest	request object	None
id	int	group id	None

Returns:

Type	Description
HttpResponse	updated group as json response

View Source

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[
            _default_group_example,
            OpenApiExample(
                name="Update group partially",
                description="Update only some group fields.",
                value={"name": "My new amazing group is the best!"},
            )
        ]
    )
    def partial_update(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Update group information partially.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: updated group as json response
        """
        return self._update(request, id, partial=True)

perform_authentication

def perform_authentication(
    self,
    request
)

Perform authentication on the incoming request.

Note that if you override this and simply 'pass', then authentication will instead be performed lazily, the first time either request.user or request.auth is accessed.

View Source

    def perform_authentication(self, request):
        """
        Perform authentication on the incoming request.

        Note that if you override this and simply 'pass', then authentication
        will instead be performed lazily, the first time either
        `request.user` or `request.auth` is accessed.
        """
        request.user

perform_content_negotiation

def perform_content_negotiation(
    self,
    request,
    force=False
)

Determine which renderer and media type to use render the response.

View Source

    def perform_content_negotiation(self, request, force=False):
        """
        Determine which renderer and media type to use render the response.
        """
        renderers = self.get_renderers()
        conneg = self.get_content_negotiator()

        try:
            return conneg.select_renderer(request, renderers, self.format_kwarg)
        except Exception:
            if force:
                return (renderers[0], renderers[0].media_type)
            raise

permission_denied

def permission_denied(
    self,
    request,
    message=None,
    code=None
)

If request is not permitted, determine what kind of exception to raise.

View Source

    def permission_denied(self, request, message=None, code=None):
        """
        If request is not permitted, determine what kind of exception to raise.
        """
        if request.authenticators and not request.successful_authenticator:
            raise exceptions.NotAuthenticated()
        raise exceptions.PermissionDenied(detail=message, code=code)

raise_uncaught_exception

def raise_uncaught_exception(
    self,
    exc
)

View Source

    def raise_uncaught_exception(self, exc):
        if settings.DEBUG:
            request = self.request
            renderer_format = getattr(request.accepted_renderer, 'format')
            use_plaintext_traceback = renderer_format not in ('html', 'api', 'admin')
            request.force_plaintext_errors(use_plaintext_traceback)
        raise exc

retrieve

def retrieve(
    self,
    request: django.http.request.HttpRequest,
    id: int
) -> django.http.response.HttpResponse

Get group information by id.

Parameters:

Name	Type	Description	Default
request	HttpRequest	request object	None
id	int	group id	None

Returns:

Type	Description
HttpResponse	group as json response

View Source

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[_default_group_example]
    )
    def retrieve(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Get group information by id.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: group as json response
        """
        group = self._get_group(request.user, id)
        serializer = GroupSerializer(group)
        return Response(serializer.data)

reverse_action

def reverse_action(
    self,
    url_name,
    *args,
    **kwargs
)

Reverse the action for the given url_name.

View Source

    def reverse_action(self, url_name, *args, **kwargs):
        """
        Reverse the action for the given `url_name`.
        """
        url_name = '%s-%s' % (self.basename, url_name)
        namespace = None
        if self.request and self.request.resolver_match:
            namespace = self.request.resolver_match.namespace
        if namespace:
            url_name = namespace + ':' + url_name
        kwargs.setdefault('request', self.request)

        return reverse(url_name, *args, **kwargs)

setup

def setup(
    self,
    request,
    *args,
    **kwargs
)

Initialize attributes shared by all view methods.

View Source

    def setup(self, request, *args, **kwargs):
        """Initialize attributes shared by all view methods."""
        if hasattr(self, "get") and not hasattr(self, "head"):
            self.head = self.get
        self.request = request
        self.args = args
        self.kwargs = kwargs

throttled

def throttled(
    self,
    request,
    wait
)

If request is throttled, determine what kind of exception to raise.

View Source

    def throttled(self, request, wait):
        """
        If request is throttled, determine what kind of exception to raise.
        """
        raise exceptions.Throttled(wait)

update

def update(
    self,
    request: django.http.request.HttpRequest,
    id: int
) -> django.http.response.HttpResponse

Update group information.

Parameters:

Name	Type	Description	Default
request	HttpRequest	request object	None
id	int	group id	None

Returns:

Type	Description
HttpResponse	updated group as json response

View Source

    @extend_schema(
        responses={
            status.HTTP_200_OK: GroupSerializer,
            status.HTTP_400_BAD_REQUEST: ErrorSerializer,
            status.HTTP_403_FORBIDDEN: error_response_403,
        },
        examples=[
            _default_group_example,
            OpenApiExample(
                name="Update group",
                description="Update group fields.",
                value={"name": "My new amazing group is the best!"},
            )
        ]
    )
    def update(self, request: HttpRequest, id: int) -> HttpResponse:
        """
        Update group information.

        Args:
            request (HttpRequest): request object
            id (int): group id

        Returns:
            HttpResponse: updated group as json response
        """
        return self._update(request, id, partial=False)