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()anddefer()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:
- Use ViewSets for standard CRUD, customize with mixins
- Transform field names in serializers for API-friendly interfaces
- Implement declarative permissions with drf-access-policy
- Validate query parameters explicitly in FilterSets
- Paginate all list endpoints with custom metadata
- Version APIs using URL path versioning
- Handle errors consistently with HTTP 422 for validation
- Document ViewSets with comprehensive docstrings
- Test thoroughly with pytest and APIClient
- 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¶
- Django REST Framework Documentation
- drf-access-policy Documentation
- django-filter Documentation
- HTTP Status Codes
- REST API Design Best Practices
- Two Scoops of Django: REST APIs
This documentation reflects Django REST Framework patterns for Django 5.2+ applications (2025), based on production implementations and theoretical best practices.