💻 Примеры DRF-блока

🎯 Показательные паттерны ← К оглавлению урока

⚡ Ключевые примеры

# Агрегация
from django.db.models import Count, Avg
stats = Book.objects.aggregate(n=Count('id'), avg=Avg('price'))

# ModelSerializer + валидация
class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = '__all__'
    def validate_price(self, v):
        if v < 0: raise serializers.ValidationError("price >= 0")
        return v

# ModelViewSet + Router
class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
router.register(r'books', BookViewSet)

# query_params
author = request.query_params.get('author')
if author: qs = qs.filter(author__icontains=author)

1. Полная агрегация BookStore

Один запрос для получения общей статистики магазина книг.

from django.db.models import Count, Sum, Avg, Min, Max

# models.py
from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.CharField(max_length=100)
    published_date = models.DateField()
    price = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
    discounted_price = models.DecimalField(max_digits=10, decimal_places=2, null=True, blank=True)
    is_bestseller = models.BooleanField(default=False)

    def __str__(self):
        return self.title

# Общая статистика магазина
stats = Book.objects.aggregate(
    total_books=Count('id'),
    bestsellers=Count('id', filter=Q(is_bestseller=True)),
    avg_price=Avg('price'),
    total_value=Sum('price'),
    cheapest=Min('price'),
    most_expensive=Max('price'),
)
print(stats)
# {'total_books': 50, 'bestsellers': 12, 'avg_price': 19.5, ...}

2. Статистика по авторам через annotate

from django.db.models import Count, Avg

# Количество книг и средняя цена для каждого автора
author_stats = (
    Book.objects
    .values('author')
    .annotate(
        book_count=Count('id'),
        avg_price=Avg('price'),
    )
    .order_by('-book_count')
)

for entry in author_stats:
    print(f"{entry['author']}: {entry['book_count']} книг, "
          f"средняя цена {entry['avg_price']:.2f}")

3. Подзапрос: книги ниже средней цены своего издательства

from django.db.models import OuterRef, Subquery, Avg

# Подзапрос: средняя цена книг в том же издательстве
avg_price_subquery = (
    Book.objects
    .filter(publisher=OuterRef('publisher'))
    .values('publisher')
    .annotate(avg_p=Avg('price'))
    .values('avg_p')
)

# Книги с ценой ниже средней для их издательства
cheap_books = Book.objects.filter(
    price__lt=Subquery(avg_price_subquery)
).select_related('publisher')

for book in cheap_books:
    print(book.title, book.price, book.publisher.name)

4. Полный CRUD API (объединённые маршруты)

# serializers.py
from rest_framework import serializers
from .models import Book

class BookListSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = ['id', 'title', 'author', 'price']

class BookDetailSerializer(serializers.ModelSerializer):
    created_at = serializers.DateTimeField(read_only=True)
    class Meta:
        model = Book
        fields = '__all__'

# views.py
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework import status
from .models import Book
from .serializers import BookListSerializer, BookDetailSerializer

@api_view(['GET', 'POST'])
def book_list_create(request):
    if request.method == 'GET':
        books = Book.objects.all()
        serializer = BookListSerializer(books, many=True)
        return Response(serializer.data, status=status.HTTP_200_OK)

    elif request.method == 'POST':
        serializer = BookDetailSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['GET', 'PUT', 'DELETE'])
def book_detail_update_delete(request, pk):
    try:
        book = Book.objects.get(pk=pk)
    except Book.DoesNotExist:
        return Response({'error': 'Book not found'}, status=status.HTTP_404_NOT_FOUND)

    if request.method == 'GET':
        serializer = BookDetailSerializer(book)
        return Response(serializer.data)

    elif request.method == 'PUT':
        serializer = BookDetailSerializer(book, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    elif request.method == 'DELETE':
        book.delete()
        return Response({'message': 'Deleted'}, status=status.HTTP_204_NO_CONTENT)

# urls.py
from django.urls import path
from .views import book_list_create, book_detail_update_delete

urlpatterns = [
    path('books/', book_list_create, name='book-list-create'),
    path('books/<int:pk>/', book_detail_update_delete, name='book-detail'),
]

5. ModelViewSet + Router (полный CRUD за 10 строк)

# views.py
from rest_framework import viewsets
from .models import Book
from .serializers import BookDetailSerializer

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all().order_by('-published_date')
    serializer_class = BookDetailSerializer

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

router = DefaultRouter()
router.register(r'books', BookViewSet, basename='book')
urlpatterns = router.urls

6. Динамическая фильтрация + пагинация через query_params

from rest_framework.views import APIView
from rest_framework.response import Response
from .models import Book
from .serializers import BookListSerializer

class BookSearchView(APIView):
    def get(self, request):
        qs = Book.objects.all()

        # Фильтры
        author = request.query_params.get('author')
        is_bestseller = request.query_params.get('is_bestseller')
        min_price = request.query_params.get('min_price')
        max_price = request.query_params.get('max_price')

        if author:
            qs = qs.filter(author__icontains=author)
        if is_bestseller in ('true', '1'):
            qs = qs.filter(is_bestseller=True)
        if min_price:
            qs = qs.filter(price__gte=min_price)
        if max_price:
            qs = qs.filter(price__lte=max_price)

        # Сортировка
        ordering = request.query_params.get('ordering', '-published_date')
        allowed = ['title', '-title', 'price', '-price', 'published_date', '-published_date']
        if ordering in allowed:
            qs = qs.order_by(ordering)

        # Пагинация
        page = max(1, int(request.query_params.get('page', 1)))
        page_size = min(100, int(request.query_params.get('page_size', 10)))
        start = (page - 1) * page_size
        qs = qs[start:start + page_size]

        serializer = BookListSerializer(qs, many=True)
        return Response({
            'page': page,
            'page_size': page_size,
            'results': serializer.data,
        })

7. Task API — мини-проект из ДЗ

# models.py
from django.db import models

class Task(models.Model):
    STATUS_CHOICES = [
        ('todo', 'To Do'),
        ('in_progress', 'In Progress'),
        ('done', 'Done'),
    ]
    title = models.CharField(max_length=200)
    description = models.TextField(blank=True)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='todo')
    deadline = models.DateField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.title

# serializers.py
from rest_framework import serializers
from .models import Task

class TaskSerializer(serializers.ModelSerializer):
    created_at = serializers.DateTimeField(read_only=True)
    class Meta:
        model = Task
        fields = '__all__'

# views.py
from django.db.models import Count
from django.utils import timezone
from rest_framework.decorators import api_view
from rest_framework.response import Response
from rest_framework import status
from .models import Task
from .serializers import TaskSerializer

@api_view(['GET', 'POST'])
def task_list_create(request):
    if request.method == 'GET':
        tasks = Task.objects.all()
        serializer = TaskSerializer(tasks, many=True)
        return Response(serializer.data)
    elif request.method == 'POST':
        serializer = TaskSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['GET'])
def task_stats(request):
    today = timezone.now().date()
    stats = Task.objects.aggregate(
        total=Count('id'),
        todo=Count('id', filter=Q(status='todo')),
        in_progress=Count('id', filter=Q(status='in_progress')),
        done=Count('id', filter=Q(status='done')),
        overdue=Count('id', filter=Q(deadline__lt=today) & ~Q(status='done')),
    )
    return Response(stats)