Skip to content

Django REST Framework API Patterns

This guide documents Django REST Framework (DRF) patterns for building production-grade APIs with Django 5.2+ (2025). These patterns are derived from real-world production applications and represent theoretical best practices for scalable, maintainable REST APIs.

Overview

Django REST Framework extends Django to provide powerful tools for building Web APIs. It offers serialization, authentication, permissions, filtering, pagination, and comprehensive API documentation capabilities while maintaining Django's philosophy of explicit, maintainable code.

Philosophy

DRF follows Django's principles: explicit over implicit, DRY, and loose coupling. Well-designed APIs prioritize consistency, predictability, and developer experience.

ViewSet Patterns

ViewSets combine the logic for handling multiple related views into a single class. They provide standard CRUD operations with minimal code while remaining customizable for complex requirements.

Basic ViewSet Structure

# views/api/people/planion_people.py
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from django_filters.rest_framework import DjangoFilterBackend

from .permissions import PeopleAccessPolicy
from poseidon.commons.mixins.account_validation import AccountValidationMixin
from poseidon.commons.pagination.drf_pagination import CustomPageNumberPagination
from poseidon.serializers.planion_people import PlanionPeopleSerializer
from poseidon.models.planion.planion_people import People


class PlanionPeopleViewSet(AccountValidationMixin, viewsets.ModelViewSet):
    """
    ViewSet for managing Planion people records.

    Provides full CRUD operations with filtering, pagination, and account-based authorization.
    """

    serializer_class = PlanionPeopleSerializer
    permission_classes = [IsAuthenticated, PeopleAccessPolicy]
    filter_backends = [DjangoFilterBackend]
    filterset_class = PeopleFilter
    pagination_class = CustomPageNumberPagination
    http_method_names = ["get", "post", "put", "patch", "delete"]

    def get_queryset(self):
        """Return queryset filtered by account."""
        account = self.get_account(self.request)
        return People.objects.using(f"{account.upper()}_RO").all()

    def perform_create(self, serializer):
        """Save new instance with account context."""
        account = self.get_account(self.request)
        serializer.save(using=f"{account.upper()}")

    def perform_update(self, serializer):
        """Update instance with account context."""
        account = self.get_account(self.request)
        serializer.save(using=f"{account.upper()}")

ReadOnly ViewSet

For read-only APIs, use ReadOnlyModelViewSet to restrict mutations:

# views/api/container/planion_container.py
from rest_framework import viewsets

class PlanionContainerViewSet(AccountValidationMixin, viewsets.ReadOnlyModelViewSet):
    """
    Read-only ViewSet for container data.

    Provides list and retrieve operations only.
    """

    serializer_class = PlanionContainerSerializer
    permission_classes = [IsAuthenticated, ContainerAccessPolicy]
    filter_backends = [DjangoFilterBackend]
    filterset_class = ContainerFilter
    pagination_class = CustomPageNumberPagination
    http_method_names = ["get"]
    queryset = PlanionContainer.objects.all().order_by("code")

Generic ViewSet with Mixins

Compose ViewSets from mixins for fine-grained control:

from rest_framework import mixins, viewsets

class CustomViewSet(
    mixins.CreateModelMixin,
    mixins.RetrieveModelMixin,
    mixins.ListModelMixin,
    viewsets.GenericViewSet,
):
    """
    ViewSet with create, retrieve, and list operations only.

    Omits update and delete operations.
    """

    serializer_class = MySerializer
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        """Custom queryset with optimizations."""
        return MyModel.objects.select_related("related_model").all()

ViewSet with Custom Actions

Add custom endpoints beyond standard CRUD:

from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework import status

class SessionViewSet(viewsets.ReadOnlyModelViewSet):
    """ViewSet for session management."""

    serializer_class = SessionSerializer

    @action(detail=True, methods=["post"])
    def embargo_lift(self, request, pk=None):
        """
        Lift embargo on a session.

        Custom action: POST /api/v1/sessions/{id}/embargo_lift/
        """
        session = self.get_object()
        session.embargo_lifted = True
        session.embargo_lift_datetime = timezone.now()
        session.save()

        return Response(
            {"status": "embargo lifted", "session_id": session.id},
            status=status.HTTP_200_OK
        )

    @action(detail=False, methods=["get"])
    def upcoming(self, request):
        """
        Get upcoming sessions.

        Custom action: GET /api/v1/sessions/upcoming/
        """
        queryset = self.get_queryset().filter(
            start_time__gte=timezone.now()
        ).order_by("start_time")[:10]

        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

Overriding ViewSet Methods

Customize behavior by overriding ViewSet methods:

class PlanionScheduleViewSet(viewsets.ModelViewSet):
    """ViewSet with custom list and retrieve behavior."""

    def list(self, request, *args, **kwargs):
        """Custom list implementation with additional context."""
        queryset = self.filter_queryset(self.get_queryset())
        page = self.paginate_queryset(queryset)

        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

    def retrieve(self, request, *args, **kwargs):
        """Custom retrieve with embargo rule execution."""
        instance = self.get_object()

        # Run custom business logic
        self.execute_embargo_rules(instance)

        serializer = self.get_serializer(instance)
        return Response(serializer.data)

Serializer Organization

Serializers transform Django models to/from JSON and handle validation. Organize serializers by domain or feature for maintainability.

Model Serializer with Field Mapping

Transform database field names to API-friendly names:

# serializers/planion_people.py
from rest_framework import serializers
from poseidon.models.planion.planion_people import People


class PlanionPeopleSerializer(serializers.ModelSerializer):
    """
    Serializer for People model.

    Maps database field names to user-friendly API field names.
    """

    created_at = serializers.DateTimeField()
    updated_at = serializers.DateTimeField()
    planstone_pid = serializers.IntegerField(source="pid")
    client_pid = serializers.CharField(source="clientpid")
    client_pid2 = serializers.CharField(source="clientpid2")
    address_1 = serializers.CharField(source="add1")
    address_2 = serializers.CharField(source="add2")
    biography = serializers.CharField(source="bio")
    first_name = serializers.CharField(source="fname")
    last_name = serializers.CharField(source="lname")
    middle_initial = serializers.CharField(source="mi")
    email = serializers.CharField()
    email_2 = serializers.CharField(source="email2")
    email_3 = serializers.CharField(source="email3")
    phone = serializers.CharField(source="dayphone")
    phone_work = serializers.CharField(source="workphone")
    is_test_account = serializers.BooleanField(source="testaccnt")

    class Meta:
        model = People
        fields = [
            "created_at",
            "updated_at",
            "planstone_pid",
            "client_pid",
            "client_pid2",
            "address_1",
            "address_2",
            "biography",
            "first_name",
            "last_name",
            "middle_initial",
            "email",
            "email_2",
            "email_3",
            "phone",
            "phone_work",
            "is_test_account",
        ]

Field Naming Convention

Always use snake_case for API field names, not camelCase. This aligns with Python conventions and DRF standards.

Serializer with Computed Fields

Use SerializerMethodField for computed or derived values:

# serializers/serializer_planion_sessions.py
from rest_framework import serializers
from datetime import datetime, timedelta


class ScheduleSerializer(serializers.ModelSerializer):
    """
    Serializer for Schedule model with computed fields.
    """

    schedule_id = serializers.IntegerField(source="id")
    session_title = serializers.CharField(source="session.title", read_only=True)
    start_time = serializers.SerializerMethodField()
    end_time = serializers.SerializerMethodField()
    duration_mins = serializers.SerializerMethodField()
    room_name = serializers.SerializerMethodField()
    session_published = serializers.SerializerMethodField()
    speakers = serializers.SerializerMethodField()

    class Meta:
        model = Schedule
        fields = [
            "schedule_id",
            "session_title",
            "start_time",
            "end_time",
            "duration_mins",
            "room_name",
            "session_published",
            "speakers",
        ]
        read_only_fields = fields

    def get_session_published(self, obj):
        """Convert published field to boolean."""
        return bool(obj.session.published)

    def get_duration_mins(self, obj):
        """Calculate duration in minutes."""
        if obj.session.durmins == 0 or obj.session.durmins is None:
            if obj.rstartmins is not None and obj.rendmins is not None:
                return int(obj.rendmins) - int(obj.rstartmins)
            return None
        return obj.session.durmins

    def get_start_time(self, obj):
        """Convert date and decimal minutes to datetime."""
        if obj.rday and obj.rstartmins is not None:
            return self.convert_to_datetime(obj.rday, obj.rstartmins)
        return None

    def get_end_time(self, obj):
        """Convert date and decimal minutes to datetime."""
        if obj.rday and obj.rendmins is not None:
            return self.convert_to_datetime(obj.rday, obj.rendmins)
        return None

    def convert_to_datetime(self, date, minutes):
        """Convert decimal minutes to datetime."""
        minutes_as_int = int(minutes)
        return datetime.combine(date, datetime.min.time()) + timedelta(minutes=minutes_as_int)

    def get_speakers(self, obj):
        """Serialize related speakers using prefetched data."""
        speakers = obj.pslink_session.all()
        return PlanionPslinkSerializer(speakers, many=True, context=self.context).data

    def get_room_name(self, obj):
        """Get room name with caching optimization."""
        account = self.context.get("account")

        # Use instance-level cache for batch processing
        if not hasattr(self, "_cached_rooms"):
            room_ids = [
                sched.roomid for sched in self.instance
                if sched.roomid and sched.roomid not in ["PARK", "NOFORM"]
            ]
            rooms = Rooms.objects.using(f"{account.upper()}_RO").filter(
                id__in=room_ids
            ).select_related("hotel")
            self._cached_rooms = {room.id: room for room in rooms}

        room = self._cached_rooms.get(obj.roomid)
        return room.roomname if room else obj.roomname

Nested Serializers

Handle related objects with nested serializers:

# serializers/planion_forms.py
from rest_framework import serializers


class PlanionEvalDefSerializer(serializers.ModelSerializer):
    """Serializer for form questions."""

    question_id = serializers.CharField(source="id")
    question_text = serializers.CharField(source="webdesc")
    field_type = serializers.CharField(source="fieldtype")

    class Meta:
        model = Evaldef
        fields = ["question_id", "question_text", "field_type"]


class PlanionEvalFormSerializer(serializers.ModelSerializer):
    """
    Serializer for forms with nested questions.
    """

    form_id = serializers.CharField(source="formid")
    form_description = serializers.CharField(source="formdesc")
    container = serializers.CharField(source="confx")
    collection_level = serializers.SerializerMethodField()
    questions = PlanionEvalDefSerializer(many=True, read_only=True, source="evaldefs")

    class Meta:
        model = Evalform
        fields = [
            "created_at",
            "updated_at",
            "form_id",
            "container",
            "form_description",
            "collection_level",
            "questions",
        ]

    def get_collection_level(self, obj):
        """Map context to human-readable label."""
        if not obj.context:
            return None

        context_mapping = {
            "CONF": "CONTAINER",
            "SCHEDID": "SESSION",
            "PSLINKID": "PRESENTATION",
            "DOCID": "DOCUMENT",
        }

        return context_mapping.get(obj.context, obj.context)

Read-Only vs Writable Fields

Control field mutability explicitly:

class MySerializer(serializers.ModelSerializer):
    """Serializer with mixed read/write fields."""

    # Read-only computed field
    full_name = serializers.SerializerMethodField()

    # Writable field with validation
    email = serializers.EmailField(required=True)

    # Read-only relationship
    created_by = serializers.StringRelatedField(read_only=True)

    # Write-only sensitive field
    password = serializers.CharField(write_only=True, min_length=8)

    class Meta:
        model = MyModel
        fields = ["full_name", "email", "created_by", "password", "status"]
        read_only_fields = ["status"]

    def get_full_name(self, obj):
        """Computed full name."""
        return f"{obj.first_name} {obj.last_name}"

    def create(self, validated_data):
        """Override create to handle write-only fields."""
        password = validated_data.pop("password")
        instance = MyModel.objects.create(**validated_data)
        instance.set_password(password)
        instance.save()
        return instance

Permission Classes (drf-access-policy)

Django REST Framework uses drf-access-policy for declarative, reusable permission management. This approach separates authorization logic from views and enables role-based access control.

Basic Access Policy

# views/api/people/permissions.py
from rest_access_policy import AccessPolicy
from poseidon.commons.mixins.account_authorization import AccountAuthorizationMixin


class PeopleAccessPolicy(AccountAuthorizationMixin, AccessPolicy):
    """
    Permission policy for People API.

    Defines who can perform which actions based on Django groups.
    """

    statements = [
        {
            "action": ["list", "retrieve"],
            "principal": ["group:API | people", "group:pytest_runner"],
            "effect": "allow",
            "condition": "has_account_authorization",
        },
        {
            "action": ["create", "update", "partial_update", "destroy"],
            "principal": [
                "group:API | people",
                "group:API | people_rw",
                "group:pytest_runner",
            ],
            "effect": "allow",
            "condition": "has_account_authorization",
        },
    ]

Access Policy with Custom Conditions

Implement custom authorization conditions:

# views/api/session/permissions.py
from rest_access_policy import AccessPolicy
from poseidon.commons.mixins.account_authorization import AccountAuthorizationMixin


class SessionAccessPolicy(AccountAuthorizationMixin, AccessPolicy):
    """
    Permission policy for Session API with custom conditions.
    """

    statements = [
        {
            "action": ["list", "retrieve"],
            "principal": ["group:API | session", "group:pytest_runner"],
            "effect": "allow",
            "condition": "has_account_authorization",
        },
        {
            "action": ["embargo_lift"],
            "principal": ["group:API | session_admin"],
            "effect": "allow",
            "condition": ["has_account_authorization", "is_embargo_eligible"],
        },
    ]

    def is_embargo_eligible(self, request, view, action) -> bool:
        """Custom condition: check if session is embargo eligible."""
        if action != "embargo_lift":
            return True

        session = view.get_object()
        return bool(getattr(session, "embeligible", False))

Account Authorization Mixin

Multi-tenant authorization pattern:

# commons/mixins/account_authorization.py
from django.conf import settings
from django.http import Http404
import logging


logger = logging.getLogger(__name__)


class AccountAuthorizationMixin:
    """
    Mixin for account-based authorization.

    Validates user has access to requested account.
    """

    def has_account_authorization(self, request, view, action) -> bool:
        """
        Check if user has authorization for the requested account.

        Returns:
            bool: True if authorized, False otherwise
        """
        try:
            account_name = self.get_account(request)
        except Http404 as e:
            logger.warning(f"Authorization failed: {e}")
            return False

        group = f"account_access_{account_name}"
        user = request.user

        # Check if pytest runner group is allowed via settings
        if (
            getattr(settings, "ALLOW_PYTEST_BYPASS", False)
            and user.groups.filter(name="pytest_runner").exists()
        ):
            return True

        # Check if the user is a superuser
        if user.is_superuser:
            return True

        # Check if the user is an admin
        if user.is_staff:
            return True

        # Check if the user belongs to the required group
        if user.groups.filter(name=group).exists():
            return True

        logger.info(
            f"Authorization denied: User '{user.username}' does not have access to account '{account_name}'."
        )
        return False

Simple Access Policy

For straightforward authorization:

# views/api/base/permissions.py
from rest_access_policy import AccessPolicy


class HelloWorldAccessPolicy(AccessPolicy):
    """
    Simple access policy for hello world endpoint.
    """

    statements = [
        {
            "action": ["hello_world"],
            "principal": ["group:API | hello_world", "group:pytest_runner"],
            "effect": "allow",
        },
    ]

Authentication

Django REST Framework supports multiple authentication methods. Production applications typically use token-based or JWT authentication for APIs.

Bearer Token Authentication

Extend DRF's TokenAuthentication to use Bearer tokens:

# models/bearer_authentication.py
from rest_framework.authentication import TokenAuthentication


class BearerAuthentication(TokenAuthentication):
    """
    Token authentication using 'Bearer' keyword.

    Clients include: Authorization: Bearer <token>
    """

    keyword = "Bearer"

Configure in settings:

# settings/base.py
REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": (
        "poseidon.models.BearerAuthentication",
    ),
    "DEFAULT_PERMISSION_CLASSES": (
        "rest_framework.permissions.IsAuthenticated",
    ),
}

Function-Based View Authentication

Apply authentication to function-based views:

# views/api/base/methods.py
import jwt
from datetime import datetime, timedelta
from django.conf import settings
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response


@api_view(["GET"])
@permission_classes([IsAuthenticated])
def hello_world(request):
    """
    Simple authenticated endpoint.

    Returns:
        Response: Current server time
    """
    return Response({"message": datetime.now()})


@api_view(["POST"])
@permission_classes([IsAuthenticated])
def client_access_token(request):
    """
    Generate client access token.

    Returns:
        Response: JWT token with 4-hour expiration
    """
    token = jwt.encode(
        {
            "sub": "none",
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(hours=4),
        },
        settings.GENERAL_TOKEN_SECRET,
        algorithm="HS256",
    )

    return Response({"token": token})

OAuth2 and Social Authentication

Configure social authentication providers:

# settings/base.py
INSTALLED_APPS = [
    # ...
    "social_django",
    # ...
]

# Auth0 Configuration
SOCIAL_AUTH_AUTH0_DOMAIN = "your-domain.auth0.com"
SOCIAL_AUTH_AUTH0_SCOPE = ["openid", "profile"]
SOCIAL_AUTH_TRAILING_SLASH = False

# OAuth2 Settings
if SOCIAL_AUTH_AUTH0_DOMAIN:
    AUDIENCE = f"https://{SOCIAL_AUTH_AUTH0_DOMAIN}/userinfo"
    SOCIAL_AUTH_AUTH0_AUTH_EXTRA_ARGUMENTS = {"audience": AUDIENCE}
# settings/production.py
from .base import *

# Load OAuth credentials from parameter store
SOCIAL_AUTH_AUTH0_KEY = pconfig.get_param("SOCIAL_AUTH_AUTH0_KEY")
SOCIAL_AUTH_AUTH0_SECRET = pconfig.get_param("SOCIAL_AUTH_AUTH0_SECRET")
SOCIAL_AUTH_REDIRECT_IS_HTTPS = True

JWT Configuration

For JWT-based authentication:

# settings/base.py
SIMPLE_JWT = {
    "USE_JWT": False,  # Set to True to enable
    "ACCESS_TOKEN_LIFETIME": timedelta(hours=4),
    "REFRESH_TOKEN_LIFETIME": timedelta(days=7),
    "ROTATE_REFRESH_TOKENS": True,
    "ALGORITHM": "HS256",
    "SIGNING_KEY": SECRET_KEY,
}

Filtering and Search (django-filter)

Django-filter provides declarative filtering for querysets. Define custom FilterSets to validate and process query parameters.

Basic FilterSet

# views/api/people/planion_people.py
import datetime
from django.db.models import Q
from django_filters import rest_framework as filters
from poseidon.commons.exceptions.api import BadRequestException


class PeopleFilter(filters.FilterSet):
    """
    FilterSet for People API.

    Validates and processes query parameters.
    """

    updated_after = filters.DateTimeFilter(field_name="updated_at", lookup_expr="gte")
    any_email = filters.CharFilter(method="filter_by_any_email")
    account = filters.CharFilter(method="filter_by_account")

    class Meta:
        model = People
        fields = ["account", "pid", "clientpid", "clientpid2", "lname", "email", "username"]

    def __init__(self, *args, **kwargs):
        """Validate provided filters against allowed filters."""
        super().__init__(*args, **kwargs)

        allowed_filters = set(self.get_filters().keys())
        non_model_fields = {"updated_after", "any_email", "page", "account"}
        allowed_filters.update(non_model_fields)

        provided_filters = set(self.data.keys())

        invalid_filters = provided_filters - allowed_filters
        if invalid_filters:
            raise BadRequestException(f"Invalid filter(s): {', '.join(invalid_filters)}")

        # Convert to naive datetime since we have use_tz = False
        updated_after = self.data.get("updated_after")
        if updated_after:
            try:
                updated_after = updated_after.rstrip("Z")
                naive_datetime = datetime.datetime.strptime(
                    updated_after, "%Y-%m-%dT%H:%M:%S"
                )
                self.data = self.data.copy()
                self.data["updated_after"] = naive_datetime
            except ValueError:
                raise BadRequestException(f"Invalid updated_after: {updated_after}")

    def filter_by_any_email(self, queryset, name, value):
        """Filter by any of the three email fields."""
        return queryset.filter(
            Q(email__iexact=value) | Q(email2__iexact=value) | Q(email3__iexact=value)
        )

    def filter_by_account(self, queryset, name, value):
        """Validate account parameter without filtering."""
        return queryset

Complex FilterSet with Validation

# views/api/session/viewset_planion_session.py
import datetime
from django_filters import rest_framework as filters
import django_filters
from poseidon.commons.exceptions.api import BadRequestException


class ScheduleFilter(django_filters.FilterSet):
    """
    FilterSet for Schedule API with comprehensive validation.
    """

    schedule_id = django_filters.NumberFilter(field_name="id")
    updated_after = filters.DateTimeFilter(field_name="updated_at", lookup_expr="gte")

    class Meta:
        model = Schedule
        fields = ["schedule_id"]

    def __init__(self, *args, **kwargs):
        """Validate filters and process datetime parameters."""
        super().__init__(*args, **kwargs)

        allowed_filters = set(self.filters.keys())
        non_model_fields = {"page", "account", "container"}
        allowed_filters.update(non_model_fields)

        provided_filters = set(self.data.keys())

        invalid_filters = provided_filters - allowed_filters
        if invalid_filters:
            raise BadRequestException(f"Invalid filter(s): {', '.join(invalid_filters)}")

        # Convert to naive datetime since we have use_tz = False
        updated_after = self.data.get("updated_after")
        if updated_after:
            try:
                # Remove the 'Z' and parse the datetime string
                updated_after = updated_after.rstrip("Z")
                naive_datetime = datetime.datetime.strptime(
                    updated_after, "%Y-%m-%dT%H:%M:%S"
                )
                # Add the naive datetime back to the data
                self.data = self.data.copy()
                self.data["updated_after"] = naive_datetime
            except ValueError:
                raise BadRequestException(f"Invalid updated_after: {updated_after}")

ViewSet with FilterSet

Integrate FilterSet with ViewSet:

# views/api/container/planion_container.py
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import viewsets


class PlanionContainerViewSet(AccountValidationMixin, viewsets.ModelViewSet):
    """
    ViewSet with django-filter integration.
    """

    serializer_class = PlanionContainerSerializer
    permission_classes = [IsAuthenticated, ContainerAccessPolicy]
    filter_backends = [DjangoFilterBackend]
    filterset_class = ContainerFilter
    pagination_class = CustomPageNumberPagination
    http_method_names = ["get"]
    queryset = PlanionContainer.objects.all().order_by("code")

Pagination Strategies

Pagination improves API performance and user experience by returning manageable result sets.

Custom Page Number Pagination

# commons/pagination/drf_pagination.py
from rest_framework.pagination import PageNumberPagination
from rest_framework.response import Response


class CustomPageNumberPagination(PageNumberPagination):
    """
    Custom pagination class with metadata.

    Adds links, count, and total_pages to response for better client experience.
    """

    page_size = 100
    page_size_query_param = "page_size"
    max_page_size = 1000

    def get_paginated_response(self, data):
        """Return paginated response with comprehensive metadata."""
        return Response(
            {
                "links": {
                    "next": self.get_next_link(),
                    "previous": self.get_previous_link(),
                },
                "count": self.page.paginator.count,
                "total_pages": self.page.paginator.num_pages,
                "results": data,
            }
        )


class SessionCustomPageNumberPagination(CustomPageNumberPagination):
    """Session-specific pagination with smaller page size."""

    page_size = 10

Global Pagination Configuration

# settings/base.py
REST_FRAMEWORK = {
    "DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
    "PAGE_SIZE": 100,
}

ViewSet-Level Pagination

Override pagination per ViewSet:

class MyViewSet(viewsets.ModelViewSet):
    """ViewSet with custom pagination."""

    pagination_class = SessionCustomPageNumberPagination

    def get_queryset(self):
        """Return optimized queryset for pagination."""
        return MyModel.objects.select_related("related").all()

API Versioning

API versioning maintains backward compatibility while evolving APIs.

URL Path Versioning

Recommended approach for clarity and simplicity:

# urls.py
from django.urls import include, path
from poseidon.views.api.urls import api_v1_patterns, api_v2_patterns

urlpatterns = [
    path("api/v1/", include((api_v1_patterns, "api_v1"), namespace="api_v1")),
    path("api/v2/", include((api_v2_patterns, "api_v2"), namespace="api_v2")),
]
# views/api/urls.py
from django.urls import path
from .people.planion_people import PlanionPeopleViewSet
from .container.planion_container import PlanionContainerViewSet

api_v1_patterns = [
    path("people/", PlanionPeopleViewSet.as_view({"get": "list"}), name="people-list"),
    path("people/<str:pid>/", PlanionPeopleViewSet.as_view({"get": "retrieve"}), name="people-detail"),
    path("containers/", PlanionContainerViewSet.as_view({"get": "list"}), name="container-list"),
]

api_v2_patterns = [
    # V2 endpoints with updated serializers or behavior
    path("people/", PlanionPeopleViewSetV2.as_view({"get": "list"}), name="people-list"),
]

Namespace-Based Versioning

Use Django URL namespaces:

# ViewSet remains the same
class PlanionPeopleViewSet(viewsets.ModelViewSet):
    pass

# URL configuration
urlpatterns = [
    path("api/v1/", include("myapp.api.v1.urls", namespace="v1")),
    path("api/v2/", include("myapp.api.v2.urls", namespace="v2")),
]

# In templates or views
reverse("v1:people-list")
reverse("v2:people-list")

Error Handling and Validation

Consistent error handling improves API reliability and developer experience.

Custom API Exceptions

Use HTTP 422 for validation errors:

# commons/exceptions/api.py
from rest_framework.exceptions import APIException


class BadRequestException(APIException):
    """
    Exception for validation errors.

    Uses HTTP 422 (Unprocessable Entity) instead of 400.
    """

    status_code = 422
    default_detail = "Invalid filter parameter."
    default_code = "unprocessable_entity"

Exception Usage in Views

from poseidon.commons.exceptions.api import BadRequestException

class MyViewSet(viewsets.ModelViewSet):
    def get_queryset(self):
        """Validate required parameters."""
        container = self.request.query_params.get("container")

        if not container:
            raise BadRequestException("The 'container' parameter is required.")

        return MyModel.objects.filter(container=container)

Validation in Serializers

class MySerializer(serializers.ModelSerializer):
    """Serializer with field-level and object-level validation."""

    email = serializers.EmailField()
    age = serializers.IntegerField()

    def validate_age(self, value):
        """Validate age field."""
        if value < 0:
            raise serializers.ValidationError("Age cannot be negative.")
        if value > 120:
            raise serializers.ValidationError("Age must be realistic.")
        return value

    def validate(self, data):
        """Object-level validation."""
        if data.get("start_date") > data.get("end_date"):
            raise serializers.ValidationError(
                "Start date must be before end date."
            )
        return data

Global Exception Handler

Customize exception responses:

# commons/exceptions/handlers.py
from rest_framework.views import exception_handler
from rest_framework.response import Response


def custom_exception_handler(exc, context):
    """
    Custom exception handler for consistent error responses.
    """
    # Call DRF's default handler first
    response = exception_handler(exc, context)

    if response is not None:
        # Customize error response format
        custom_response = {
            "error": {
                "status_code": response.status_code,
                "message": response.data.get("detail", "An error occurred"),
                "errors": response.data if isinstance(response.data, dict) else {},
            }
        }
        response.data = custom_response

    return response

Configure in settings:

# settings/base.py
REST_FRAMEWORK = {
    "EXCEPTION_HANDLER": "poseidon.commons.exceptions.handlers.custom_exception_handler",
}

Throttling and Rate Limiting

Protect APIs from abuse with rate limiting.

Global Throttling Configuration

# settings/base.py
REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": [
        "rest_framework.throttling.AnonRateThrottle",
        "rest_framework.throttling.UserRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "anon": "100/day",
        "user": "2500/minute",
    },
}

Custom Throttle Class

# commons/throttling.py
from rest_framework.throttling import UserRateThrottle


class BurstRateThrottle(UserRateThrottle):
    """
    Burst rate throttle for short-term limits.
    """

    scope = "burst"


class SustainedRateThrottle(UserRateThrottle):
    """
    Sustained rate throttle for long-term limits.
    """

    scope = "sustained"
# settings/base.py
REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_RATES": {
        "burst": "60/min",
        "sustained": "1000/hour",
    },
}

ViewSet-Level Throttling

from rest_framework.throttling import UserRateThrottle

class HighTrafficViewSet(viewsets.ModelViewSet):
    """ViewSet with custom throttling."""

    throttle_classes = [UserRateThrottle]
    throttle_scope = "high_traffic"

OpenAPI/Swagger Documentation

DRF generates browsable API documentation automatically.

Browsable API

Enable browsable API renderer:

# settings/base.py
REST_FRAMEWORK = {
    "DEFAULT_RENDERER_CLASSES": [
        "rest_framework.renderers.JSONRenderer",
        "rest_framework.renderers.BrowsableAPIRenderer",
    ],
}

ViewSet Docstrings

Document ViewSets and actions with docstrings:

class PlanionPeopleViewSet(viewsets.ModelViewSet):
    """
    API endpoint for managing people records.

    list:
    Return a list of all people filtered by account.

    retrieve:
    Return details of a specific person by PID.

    create:
    Create a new person record.

    update:
    Update an existing person record.

    partial_update:
    Partially update a person record.

    destroy:
    Delete a person record.
    """

    @action(detail=True, methods=["post"])
    def embargo_lift(self, request, pk=None):
        """
        Lift embargo on a person's data.

        Requires:
        - User must have embargo_admin permission
        - Person must be embargo eligible
        """
        pass

drf-spectacular Integration

For advanced OpenAPI schema generation:

# settings/base.py
INSTALLED_APPS = [
    # ...
    "drf_spectacular",
]

REST_FRAMEWORK = {
    "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema",
}

SPECTACULAR_SETTINGS = {
    "TITLE": "My API",
    "DESCRIPTION": "API documentation",
    "VERSION": "1.0.0",
    "SERVE_INCLUDE_SCHEMA": False,
}
# urls.py
from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView

urlpatterns = [
    path("api/schema/", SpectacularAPIView.as_view(), name="schema"),
    path("api/docs/", SpectacularSwaggerView.as_view(url_name="schema"), name="swagger-ui"),
]

Testing APIs

Comprehensive testing ensures API reliability and correctness.

Test Configuration

# tests/conftest.py
import pytest
from rest_framework.test import APIClient
from django.contrib.auth import get_user_model


@pytest.fixture
def api_client():
    """Provide DRF API client."""
    return APIClient()


@pytest.fixture
def authenticated_api_client(api_client, user):
    """Provide authenticated API client."""
    api_client.force_authenticate(user=user)
    return api_client


@pytest.fixture
def user(db):
    """Create test user."""
    User = get_user_model()
    return User.objects.create_user(
        username="testuser",
        email="test@example.com",
        password="testpass123",
    )


@pytest.fixture
def api_client_with_credentials(api_client, user):
    """Provide API client with bearer token."""
    from rest_framework.authtoken.models import Token

    token, _ = Token.objects.get_or_create(user=user)
    api_client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.key}")
    return api_client

Unit Tests for ViewSets

# tests/unit/api/test_people_views.py
import pytest
from django.urls import reverse
from rest_framework import status


@pytest.mark.django_db(databases=["default", "accounts", "TESTCLIENT_RO"])
def test_email_any_filter_matching_email(api_client_with_credentials, load_people_test_data):
    """
    Test filtering people by email address.

    Verifies that the any_email filter parameter correctly finds people
    with matching email addresses.
    """
    url = reverse("PeopleViewSetTOS-list")
    response = api_client_with_credentials.get(
        url, {"any_email": "mark@planstone.com", "account": "TESTCLIENT"}
    )

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1


@pytest.mark.django_db(databases=["default", "accounts", "TESTCLIENT_RO"])
def test_email_any_filter_bad_parameter(api_client_with_credentials, load_people_test_data):
    """
    Test handling of incorrect filter parameter names.

    Verifies that the API returns HTTP 422 when an incorrectly
    named filter parameter is provided.
    """
    url = reverse("PeopleViewSetTOS-list")
    response = api_client_with_credentials.get(
        url, {"email_any": "nonexistent@planstone.com", "account": "TESTCLIENT"}
    )

    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY


@pytest.mark.django_db(databases=["default", "accounts", "TESTCLIENT_RO"])
def test_filter_by_updated_after(api_client_with_credentials, load_people_test_data):
    """
    Test filtering people by last update timestamp.

    Verifies that the updated_after filter correctly finds people who were
    updated after the specified timestamp.
    """
    url = reverse("PeopleViewSetTOS-list")
    response = api_client_with_credentials.get(
        url, {"updated_after": "2024-01-02T00:00:00Z", "account": "TESTCLIENT"}
    )

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 4


@pytest.mark.django_db(databases=["default", "accounts", "TESTCLIENT_RO"])
def test_combination_of_filters(api_client_with_credentials, load_people_test_data):
    """
    Test combining multiple filters together.

    Verifies that multiple filter parameters can be combined effectively
    to narrow down search results.
    """
    url = reverse("PeopleViewSetTOS-list")
    params = {"any_email": "mark@planstone.com", "lname": "Smith", "account": "TESTCLIENT"}
    response = api_client_with_credentials.get(url, params)

    assert response.status_code == status.HTTP_200_OK
    assert len(response.data["results"]) == 1

Integration Tests

# tests/integration/api/test_container_views.py
import pytest
from django.urls import reverse
from rest_framework import status


@pytest.mark.django_db(databases=["default", "accounts", "TESTCLIENT_RO"])
def test_list_containers(api_client_with_credentials, load_container_test_data):
    """
    Test retrieving a list of containers through the API.

    Verifies that the container list endpoint returns the expected number of
    results and a 200 OK status code for authenticated requests.
    """
    url = reverse("ContainerViewSet-list")
    response = api_client_with_credentials.get(url, {"account": "TESTCLIENT"})

    assert response.status_code == status.HTTP_200_OK
    assert "results" in response.data
    assert "count" in response.data
    assert "total_pages" in response.data

Test Fixtures

# tests/conftest.py
import pytest
from django.core.management import call_command


@pytest.fixture(scope="function")
def load_people_test_data(django_db_setup, django_db_blocker):
    """
    Load people test data into the test database.

    This fixture loads the people.json fixture into the TESTCLIENT_RO database
    for use in people API endpoint tests. It clears existing data first to
    ensure test isolation across test runs.
    """
    with django_db_blocker.unblock():
        People.objects.using("TESTCLIENT_RO").all().delete()
        call_command("loaddata", "people.json", database="TESTCLIENT_RO")

Best Practices

Mixin Pattern for Shared Functionality

Use mixins to compose ViewSet behavior:

# commons/mixins/account_validation.py
from rest_framework.exceptions import NotFound
from poseidon.models.planion.planion_accounts import Accounts


class AccountValidationMixin:
    """
    Mixin to add account validation to viewsets.

    Validates account parameter exists and is active.
    """

    account_model = Accounts

    def get_account(self, request):
        """
        Check if account exists based on query params, POST data, or URL kwargs.

        Returns:
            str: Uppercase account name

        Raises:
            NotFound: If account parameter missing or account doesn't exist
        """
        # Check in the query parameters
        account_name = request.query_params.get("account")

        # If not found in query parameters, check in POST data
        if not account_name and request.method in ("POST", "PUT", "PATCH"):
            account_name = request.data.get("account")

        # If not found in POST data, check in URL kwargs
        if not account_name and "account" in request.parser_context["kwargs"]:
            account_name = request.parser_context["kwargs"]["account"]

        if not account_name:
            raise NotFound(detail="Bad Request - no account provided.")

        # Check if the account exists in the database
        account_exists = self.account_model.objects.filter(
            acronym__iexact=account_name
        ).exists()

        self.tenant_id = account_name.upper()  # Set tenant_id for queries

        if not account_exists:
            raise NotFound(detail="Bad Request - account error")

        self._account = account_name
        return account_name.upper()

Database Router Mixin

Handle multi-database operations:

# commons/mixins/account_database.py
class AccountDatabaseMixin:
    """
    Mixin for routing queries to account-specific databases.
    """

    def get_db_alias(self):
        """Get database alias for current account."""
        account = self.get_account(self.request)
        return f"{account.upper()}_RO"

    def get_queryset(self):
        """Return queryset using account-specific database."""
        db_alias = self.get_db_alias()
        return super().get_queryset().using(db_alias)

Query Optimization

Optimize queries with select_related and prefetch_related:

class PlanionScheduleViewSet(viewsets.ModelViewSet):
    """ViewSet with optimized queries."""

    def get_queryset(self):
        """Return optimized queryset."""
        from django.db.models import Prefetch

        pslink_prefetch = Prefetch(
            "pslink_session",
            queryset=Pslink.objects.order_by("sortorder").select_related(
                "abstract", "speaker"
            ),
        )

        queryset = (
            Schedule.objects.select_related("session")
            .filter(conf=F("session__conf"), session__conf=container)
            .prefetch_related(pslink_prefetch)
            .order_by("id")
        )

        return queryset

Serializer Context

Pass additional context to serializers:

class MyViewSet(viewsets.ModelViewSet):
    """ViewSet with serializer context."""

    def get_serializer_context(self):
        """Add custom context to serializer."""
        context = super().get_serializer_context()
        context["account"] = self._account
        context["container"] = self._container
        context["credit_labels"] = self._credit_labels
        return context

Consistent Field Naming

# ✅ Good: snake_case API field names
class MySerializer(serializers.ModelSerializer):
    first_name = serializers.CharField(source="fname")
    last_name = serializers.CharField(source="lname")
    created_at = serializers.DateTimeField()

# ❌ Bad: camelCase API field names
class MySerializer(serializers.ModelSerializer):
    firstName = serializers.CharField(source="fname")
    lastName = serializers.CharField(source="lname")
    createdAt = serializers.DateTimeField()

HTTP Method Restrictions

Explicitly declare allowed HTTP methods:

class ReadOnlyViewSet(viewsets.ModelViewSet):
    """ViewSet with restricted HTTP methods."""

    http_method_names = ["get", "head", "options"]

    # Or for specific operations
    http_method_names = ["get", "post", "patch", "delete"]

Error Detail Consistency

Provide helpful error messages:

# ✅ Good: Descriptive error messages
if not container:
    raise ValidationError("The 'container' parameter is required.")

if invalid_filters:
    raise BadRequestException(f"Invalid filter(s): {', '.join(invalid_filters)}")

# ❌ Bad: Generic error messages
if not container:
    raise ValidationError("Missing parameter")

Common Patterns from Poseidon

Multi-Tenant ViewSet

class PlanionFormViewSet(AccountValidationMixin, AccountDatabaseMixin, viewsets.ModelViewSet):
    """
    Multi-tenant ViewSet with account-specific database routing.
    """

    serializer_class = PlanionEvalFormSerializer
    permission_classes = [IsAuthenticated, FormsAccessPolicy]
    filter_backends = [DjangoFilterBackend]
    pagination_class = CustomPageNumberPagination
    filterset_class = EvalFormFilter
    queryset = Evalform.objects.all()

    def get_queryset(self):
        """Filter queryset by container and prefetch related data."""
        container = self.request.query_params.get("container")

        if not container:
            raise ValidationError("The 'container' parameter is required.")

        queryset = super().get_queryset()
        queryset = queryset.filter(confx=container)

        return queryset.prefetch_related(
            Prefetch(
                "evaldefs",
                queryset=Evaldef.objects.exclude(fieldtype="HEADING")
                .exclude(outfldname="CONF")
                .order_by("sortorder")
                .prefetch_related(
                    Prefetch(
                        "subfields",
                        queryset=PlanionSubfield.objects.order_by("sortorder"),
                    )
                ),
            )
        ).order_by("formid")

Dynamic Label Annotation

class PlanionScheduleViewSet(viewsets.ModelViewSet):
    """ViewSet with dynamic field label annotation."""

    def fetch_custom_field_labels(self):
        """Fetch and cache custom field labels."""
        if not hasattr(self, "_custom_field_labels"):
            labels = {}
            for i in range(1, 21):
                field_name = f"CUSTOMM{i}"
                label_subquery = Cstmflds.objects.filter(
                    conf=self._container, table="SESSION", field=field_name
                ).values("desc")[:1]
                labels[f"custom_field_{i}_label"] = Subquery(label_subquery)
            self._custom_field_labels = labels
        return self._custom_field_labels

    def get_queryset(self):
        """Annotate queryset with dynamic labels."""
        queryset = super().get_queryset()

        self._custom_field_labels = self.fetch_custom_field_labels()
        queryset = queryset.annotate(**self._custom_field_labels)

        return queryset

Performance Considerations

Query Optimization Checklist

  • Use select_related() for foreign key relationships
  • Use prefetch_related() for reverse foreign keys and many-to-many
  • Annotate computed fields at database level when possible
  • Cache expensive lookups (labels, configurations) at instance level
  • Use only() and defer() to limit field selection
  • Add database indexes for filtered fields
  • Use iterator() for large querysets
  • Implement pagination for all list endpoints

Serializer Performance

class OptimizedSerializer(serializers.ModelSerializer):
    """Serializer with performance optimizations."""

    def get_room_name(self, obj):
        """Use batch-level caching for related lookups."""
        # Cache at serializer instance level for entire result set
        if not hasattr(self, "_cached_rooms"):
            room_ids = [
                item.roomid for item in self.instance
                if item.roomid and item.roomid not in ["PARK", "NOFORM"]
            ]
            rooms = Rooms.objects.filter(id__in=room_ids).select_related("hotel")
            self._cached_rooms = {room.id: room for room in rooms}

        room = self._cached_rooms.get(obj.roomid)
        return room.roomname if room else obj.roomname

Conclusion

Django REST Framework provides a comprehensive toolkit for building production-grade APIs. Key principles for success:

  1. Use ViewSets for standard CRUD, customize with mixins
  2. Transform field names in serializers for API-friendly interfaces
  3. Implement declarative permissions with drf-access-policy
  4. Validate query parameters explicitly in FilterSets
  5. Paginate all list endpoints with custom metadata
  6. Version APIs using URL path versioning
  7. Handle errors consistently with HTTP 422 for validation
  8. Document ViewSets with comprehensive docstrings
  9. Test thoroughly with pytest and APIClient
  10. Optimize queries with select_related and prefetch_related

These patterns, derived from production Django 5.2 applications, create maintainable, scalable, and developer-friendly REST APIs.

API Design Philosophy

Good APIs are predictable, consistent, and provide clear error messages. They prioritize developer experience and maintain backward compatibility. Follow conventions, document thoroughly, and optimize for real-world usage patterns.

Further Reading


This documentation reflects Django REST Framework patterns for Django 5.2+ applications (2025), based on production implementations and theoretical best practices.