REST API Development
Build RESTful APIs using Django REST Framework for creating modern web applications and mobile apps. This is a foundational concept in Python web development that professional developers rely on daily. The explanations below are written to be beginner-friendly while covering the depth and nuance that comes from real-world Python/Django experience. Take your time with each section and practice the examples
Django REST Framework
Django REST Framework (DRF) is a powerful toolkit for building Web APIs. It provides serializers, viewsets, and authentication mechanisms for building robust APIs.. This is an essential concept that every Python/Django developer must understand thoroughly. In professional development environments, getting this right can mean the difference between code that works reliably and code that breaks in production. The following sections break this down into clear, digestible pieces with practical examples you can try immediately
API Implementation
# serializers.py
from rest_framework import serializers
from .models import Post, Category, Comment, User
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ['id', 'username', 'email', 'first_name', 'last_name', 'date_joined']
read_only_fields = ['id', 'date_joined']
class CategorySerializer(serializers.ModelSerializer):
post_count = serializers.SerializerMethodField()
class Meta:
model = Category
fields = ['id', 'name', 'slug', 'post_count']
def get_post_count(self, obj):
return obj.post_set.count()
class CommentSerializer(serializers.ModelSerializer):
author = UserSerializer(read_only=True)
class Meta:
model = Comment
fields = ['id', 'content', 'author', 'created_date']
read_only_fields = ['author', 'created_date']
class PostSerializer(serializers.ModelSerializer):
author = UserSerializer(read_only=True)
category = CategorySerializer(read_only=True)
comments = CommentSerializer(many=True, read_only=True)
comment_count = serializers.SerializerMethodField()
class Meta:
model = Post
fields = ['id', 'title', 'slug', 'content', 'excerpt', 'author',
'category', 'status', 'created_date', 'published_date',
'comments', 'comment_count']
read_only_fields = ['author', 'created_date', 'published_date']
def get_comment_count(self, obj):
return obj.comments.count()
def create(self, validated_data):
validated_data['author'] = self.context['request'].user
return super().create(validated_data)
# views.py
from rest_framework import viewsets, permissions, filters
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.pagination import PageNumberPagination
from django_filters.rest_framework import DjangoFilterBackend
class PostPagination(PageNumberPagination):
page_size = 10
page_size_query_param = 'page_size'
max_page_size = 100
class PostViewSet(viewsets.ModelViewSet):
queryset = Post.objects.all()
serializer_class = PostSerializer
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
pagination_class = PostPagination
filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
filterset_fields = ['status', 'category', 'author']
search_fields = ['title', 'content', 'excerpt']
ordering_fields = ['created_date', 'published_date', 'title']
ordering = ['-created_date']
def get_queryset(self):
queryset = Post.objects.select_related('author', 'category').prefetch_related('comments')
# Filter by status for non-admin users
if not self.request.user.is_staff:
queryset = queryset.filter(status='published')
return queryset
@action(detail=True, methods=['post'])
def like(self, request, pk=None):
post = self.get_object()
user = request.user
if user in post.likes.all():
post.likes.remove(user)
message = 'Post unliked'
else:
post.likes.add(user)
message = 'Post liked'
return Response({'message': message, 'likes_count': post.likes.count()})
@action(detail=True, methods=['post'])
def publish(self, request, pk=None):
post = self.get_object()
if request.user.has_perm('blog.change_post'):
post.status = 'published'
post.published_date = timezone.now()
post.save()
return Response({'message': 'Post published successfully'})
else:
return Response({'error': 'Permission denied'}, status=403)
class CategoryViewSet(viewsets.ReadOnlyModelViewSet):
queryset = Category.objects.all()
serializer_class = CategorySerializer
permission_classes = [permissions.AllowAny]
@action(detail=True)
def posts(self, request, pk=None):
category = self.get_object()
posts = category.post_set.filter(status='published')
serializer = PostSerializer(posts, many=True)
return Response(serializer.data)
# urls.py
from rest_framework.routers import DefaultRouter
from .views import PostViewSet, CategoryViewSet
router = DefaultRouter()
router.register(r'posts', PostViewSet)
router.register(r'categories', CategoryViewSet)
urlpatterns = router.urls
# API authentication and permissions
from rest_framework.authentication import TokenAuthentication, SessionAuthentication
from rest_framework.permissions import IsAuthenticated, IsAdminUser
class AdminPostViewSet(viewsets.ModelViewSet):
queryset = Post.objects.all()
serializer_class = PostSerializer
authentication_classes = [TokenAuthentication, SessionAuthentication]
permission_classes = [IsAdminUser]
def perform_create(self, serializer):
serializer.save(author=self.request.user)Practice Exercise: API Rate Limiting & Caching
# API Rate Limiting and Caching
from rest_framework.throttling import UserRateThrottle, AnonRateThrottle
from rest_framework.decorators import throttle_classes
from django.core.cache import cache
from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
from django.views.decorators.vary import vary_on_cookie
import hashlib
class PostRateThrottle(UserRateThrottle):
"""Custom rate limiting for post operations"""
rate = '100/hour' # 100 requests per hour for authenticated users
class AnonPostRateThrottle(AnonRateThrottle):
"""Rate limiting for anonymous users"""
rate = '20/hour' # 20 requests per hour for anonymous users
class CachedPostViewSet(viewsets.ReadOnlyModelViewSet):
"""ViewSet with caching for better performance"""
queryset = Post.objects.all()
serializer_class = PostSerializer
throttle_classes = [PostRateThrottle, AnonPostRateThrottle]
def get_cache_key(self, request):
"""Generate cache key based on request parameters"""
# Include query parameters in cache key
params = request.query_params.dict()
params_str = str(sorted(params.items()))
# Create hash of parameters
params_hash = hashlib.md5(params_str.encode()).hexdigest()
return f"posts_list_{params_hash}"
def list(self, request, *args, **kwargs):
cache_key = self.get_cache_key(request)
cached_data = cache.get(cache_key)
if cached_data is not None:
return Response(cached_data)
response = super().list(request, *args, **kwargs)
# Cache the response for 5 minutes
cache.set(cache_key, response.data, 300)
return response
@method_decorator(cache_page(60 * 15)) # Cache for 15 minutes
@method_decorator(vary_on_cookie)
def retrieve(self, request, *args, **kwargs):
return super().retrieve(request, *args, **kwargs)
# Custom API endpoints with caching
from rest_framework.decorators import api_view
from rest_framework.response import Response
@api_view(['GET'])
@throttle_classes([UserRateThrottle])
def post_statistics(request):
"""Get post statistics with caching"""
cache_key = 'post_statistics'
cached_stats = cache.get(cache_key)
if cached_stats is not None:
return Response(cached_stats)
# Calculate statistics
total_posts = Post.objects.count()
published_posts = Post.objects.filter(status='published').count()
draft_posts = Post.objects.filter(status='draft').count()
# Category statistics
category_stats = Post.objects.values('category__name').annotate(
count=Count('id')
).order_by('-count')
# Author statistics
author_stats = Post.objects.values('author__username').annotate(
count=Count('id')
).order_by('-count')[:10]
stats = {
'total_posts': total_posts,
'published_posts': published_posts,
'draft_posts': draft_posts,
'category_distribution': list(category_stats),
'top_authors': list(author_stats),
'cache_timestamp': timezone.now().isoformat()
}
# Cache for 1 hour
cache.set(cache_key, stats, 3600)
return Response(stats)
# API versioning
from rest_framework.versioning import URLPathVersioning
class VersionedPostViewSet(viewsets.ModelViewSet):
"""ViewSet with API versioning"""
versioning_class = URLPathVersioning
queryset = Post.objects.all()
def get_serializer_class(self):
"""Return different serializers based on API version"""
if self.request.version == 'v2':
return PostSerializerV2
return PostSerializer
def get_queryset(self):
"""Return different querysets based on API version"""
queryset = Post.objects.all()
if self.request.version == 'v2':
# V2 includes more fields and relationships
queryset = queryset.select_related('author', 'category').prefetch_related(
'comments', 'tags', 'likes'
)
else:
# V1 is basic
queryset = queryset.select_related('author')
return queryset
# API documentation with drf-spectacular
from drf_spectacular.utils import extend_schema, OpenApiParameter
from drf_spectacular.types import OpenApiTypes
@extend_schema(
tags=['Posts'],
summary="List all posts",
description="Retrieve a paginated list of all posts with optional filtering",
parameters=[
OpenApiParameter(
name='status',
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description='Filter posts by status (published, draft)'
),
OpenApiParameter(
name='category',
type=OpenApiTypes.INT,
location=OpenApiParameter.QUERY,
description='Filter posts by category ID'
),
],
responses={200: PostSerializer}
)
class DocumentedPostViewSet(viewsets.ModelViewSet):
"""ViewSet with comprehensive API documentation"""
queryset = Post.objects.all()
serializer_class = PostSerializer