💻 Примеры: DRF-блок (Уроки 33–38)

Пример 1: Generic Views + фильтрация + пагинация

Полноценный CRUD для задач (Task) с фильтрацией, поиском, сортировкой и пагинацией.

# views.py
from rest_framework.generics import ListCreateAPIView, RetrieveUpdateDestroyAPIView
from rest_framework.pagination import PageNumberPagination
from rest_framework import filters
from django_filters.rest_framework import DjangoFilterBackend
from .models import Task
from .serializers import TaskSerializer

class TaskPagination(PageNumberPagination):
    page_size = 10
    page_size_query_param = 'page_size'
    max_page_size = 100

class TaskListCreateView(ListCreateAPIView):
    queryset = Task.objects.all()
    serializer_class = TaskSerializer
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ['status', 'deadline']
    search_fields = ['title', 'description']
    ordering_fields = ['created_at']
    pagination_class = TaskPagination

class TaskDetailView(RetrieveUpdateDestroyAPIView):
    queryset = Task.objects.all()
    serializer_class = TaskSerializer

# urls.py
from django.urls import path
from .views import TaskListCreateView, TaskDetailView

urlpatterns = [
    path('tasks/', TaskListCreateView.as_view(), name='task-list-create'),
    path('tasks/<int:pk>/', TaskDetailView.as_view(), name='task-detail'),
]

# Примеры запросов
# GET /tasks/?status=todo
# GET /tasks/?search=important
# GET /tasks/?ordering=-created_at
# GET /tasks/?page=2&page_size=5

Пример 2: ModelViewSet + @action для подсчёта задач

# views.py
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from django.db.models import Count
from .models import Category
from .serializers import CategorySerializer

class CategoryViewSet(viewsets.ModelViewSet):
    queryset = Category.objects.all()
    serializer_class = CategorySerializer

    @action(detail=False, methods=['get'], url_path='count-tasks')
    def count_tasks(self, request):
        """Возвращает количество задач в каждой категории."""
        categories = Category.objects.annotate(task_count=Count('tasks'))
        data = [
            {'id': c.id, 'name': c.name, 'task_count': c.task_count}
            for c in categories
        ]
        return Response(data)
    # GET /categories/count-tasks/

# urls.py
from rest_framework.routers import DefaultRouter
from .views import CategoryViewSet

router = DefaultRouter()
router.register(r'categories', CategoryViewSet)
urlpatterns = [path('', include(router.urls))]

Пример 3: Soft Deletion для категорий

# managers.py
from django.db import models

class SoftDeleteManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(is_deleted=False)

# models.py
from django.utils import timezone
from .managers import SoftDeleteManager

class Category(models.Model):
    name = models.CharField(max_length=100)
    is_deleted = models.BooleanField(default=False)
    deleted_at = models.DateTimeField(null=True, blank=True)

    objects = SoftDeleteManager()

    def delete(self, *args, **kwargs):
        self.is_deleted = True
        self.deleted_at = timezone.now()
        self.save()

    def __str__(self):
        return self.name

# Тестирование в shell
# python manage.py shell
from app.models import Category

cat = Category.objects.create(name='Work')
print(Category.objects.count())  # 1

cat.delete()
print(Category.objects.count())  # 0 — SoftDeleteManager фильтрует удалённые
print(Category.objects.get(pk=cat.pk).is_deleted)  # True — физически осталось

Пример 4: Транзакция при создании связанных объектов

# views.py
from django.db import transaction
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework import status
from .models import Task, Category
from .serializers import TaskSerializer

@api_view(['POST'])
def create_task_in_category(request):
    """Создать категорию и задачу атомарно."""
    try:
        with transaction.atomic():
            category = Category.objects.create(
                name=request.data.get('category_name', 'Default')
            )
            task_data = request.data.copy()
            task_data['category'] = category.id
            serializer = TaskSerializer(data=task_data)
            serializer.is_valid(raise_exception=True)
            serializer.save()
            # Если здесь произойдёт ошибка, Category тоже откатится
            transaction.on_commit(
                lambda: print(f"Task created: {serializer.data['title']}")
            )
        return Response(serializer.data, status=status.HTTP_201_CREATED)
    except Exception as e:
        return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST)

Пример 5: get_serializer_context() для условного вывода полей

# views.py
class TaskDetailView(RetrieveAPIView):
    queryset = Task.objects.all()
    serializer_class = TaskSerializer

    def get_serializer_context(self):
        context = super().get_serializer_context()
        context['include_subtasks'] = (
            self.request.query_params.get('include_subtasks', 'false').lower() == 'true'
        )
        return context

# serializers.py
class TaskSerializer(serializers.ModelSerializer):
    class Meta:
        model = Task
        fields = '__all__'

    def to_representation(self, instance):
        representation = super().to_representation(instance)
        if self.context.get('include_subtasks'):
            representation['subtasks'] = list(
                instance.subtasks.values('id', 'title', 'status')
            )
        return representation

# GET /tasks/1/?include_subtasks=true

Пример 6: select_related + prefetch_related для оптимизации

# views.py
class TaskListView(ListAPIView):
    serializer_class = TaskSerializer

    def get_queryset(self):
        # select_related — FK/OneToOne (JOIN)
        # prefetch_related — ManyToMany (отдельный запрос)
        return Task.objects.select_related('category').prefetch_related(
            'subtasks'
        ).filter(is_deleted=False)

# Без оптимизации: 1 + N запросов (N = число задач)
# С оптимизацией: 2 запроса всегда

Пример 7: CursorPagination для стабильной навигации

# pagination.py
from rest_framework.pagination import CursorPagination

class TaskCursorPagination(CursorPagination):
    page_size = 10
    ordering = '-created_at'  # стабильное поле для курсора

# views.py
class TaskListView(ListAPIView):
    queryset = Task.objects.all()
    serializer_class = TaskSerializer
    pagination_class = TaskCursorPagination

# Ответ содержит:
# {
#   "next": "http://127.0.0.1:8000/tasks/?cursor=...",
#   "previous": null,
#   "results": [...]
# }