Шаг 09. Object-level Permissions и кастомные разрешения

🎯 Цель этапа

Аутентификация (шаг 08) отвечает на вопрос «кто ты?». Permissions отвечают на вопрос «что тебе можно?».

Наша матрица прав:

Два уровня проверки в DRF

📄 Затрагиваемые файлы

🔨 Шаги

1. Кастомные классы permissions

# apps/tasks/permissions.py
from rest_framework.permissions import BasePermission, SAFE_METHODS


class IsOwnerOrReadOnly(BasePermission):
    """
    Общий permission: безопасные методы (GET, HEAD, OPTIONS) — для всех.
    Небезопасные (POST, PUT, PATCH, DELETE) — только для «владельца» объекта.

    Владелец определяется атрибутом obj.owner (для Project) или obj.author (для Comment).
    """
    message = "Только владелец может выполнять это действие."

    def has_permission(self, request, view):
        # Читать могут все аутентифицированные пользователи
        return request.user and request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        # Безопасные методы: разрешить всем аутентифицированным
        if request.method in SAFE_METHODS:
            return True
        # Небезопасные: только владелец
        # Поддерживаем оба атрибута: owner (Project) и author (Comment)
        owner = getattr(obj, "owner", None) or getattr(obj, "author", None)
        return owner == request.user


class IsProjectMember(BasePermission):
    """
    Доступ к объекту разрешён только участникам проекта (owner + members).

    Ожидает, что объект — Task, у которого есть task.project.
    """
    message = "Вы не являетесь участником этого проекта."

    def has_permission(self, request, view):
        return request.user and request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        # obj — это Task или Comment
        # Получаем связанный проект
        project = getattr(obj, "project", None)
        if project is None:
            # Для Comment: obj.task.project
            task = getattr(obj, "task", None)
            project = getattr(task, "project", None) if task else None

        if project is None:
            return False

        return (
            request.user == project.owner
            or project.members.filter(pk=request.user.pk).exists()
        )


class IsProjectOwner(BasePermission):
    """
    Только владелец проекта может изменять/удалять сам проект.
    """
    message = "Только владелец проекта может выполнять это действие."

    def has_permission(self, request, view):
        return request.user and request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        if request.method in SAFE_METHODS:
            return True
        # obj — Project
        return obj.owner == request.user


class IsTaskAssigneeOrProjectOwner(BasePermission):
    """
    Редактировать задачу может: assignee задачи или owner проекта.
    Просматривать: любой участник проекта.
    """
    message = "Только исполнитель задачи или владелец проекта может изменять эту задачу."

    def has_permission(self, request, view):
        return request.user and request.user.is_authenticated

    def has_object_permission(self, request, view, obj):
        if request.method in SAFE_METHODS:
            # Читать может любой участник проекта
            return (
                request.user == obj.project.owner
                or obj.project.members.filter(pk=request.user.pk).exists()
            )
        # Редактировать: assignee или owner проекта
        return (
            obj.assignee == request.user
            or obj.project.owner == request.user
        )

2. Обновление ProjectViewSet — подключение permissions

# apps/projects/views.py
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated, AllowAny
from apps.tasks.permissions import IsProjectOwner
from .models import Project
from .serializers import ProjectListSerializer, ProjectDetailSerializer


class ProjectViewSet(viewsets.ModelViewSet):
    """
    CRUD для проектов.
    - list/retrieve: любой аутентифицированный пользователь
    - create: любой аутентифицированный
    - update/partial_update/destroy: только owner проекта
    """

    def get_permissions(self):
        """
        get_permissions() — динамическое назначение permissions по action.
        Вызывается DRF перед каждым запросом.
        """
        if self.action in ("update", "partial_update", "destroy"):
            # Редактировать/удалять — только owner
            return [IsAuthenticated(), IsProjectOwner()]
        # Остальные действия — требуем только аутентификацию
        return [IsAuthenticated()]

    def get_queryset(self):
        return (
            Project.objects
            .filter(is_active=True)
            .select_related("owner")
            .prefetch_related("members")
        )

    def get_serializer_class(self):
        if self.action in ("list", "create"):
            return ProjectListSerializer
        return ProjectDetailSerializer

    def perform_create(self, serializer):
        """Автоматически назначаем текущего пользователя владельцем."""
        serializer.save(owner=self.request.user)

3. Обновление TaskViewSet — подключение permissions

# apps/tasks/views.py
from rest_framework import viewsets
from rest_framework.filters import SearchFilter, OrderingFilter
from rest_framework.permissions import IsAuthenticated
from django_filters.rest_framework import DjangoFilterBackend
from .filters import TaskFilter
from .models import Task, Comment
from .permissions import IsTaskAssigneeOrProjectOwner, IsProjectMember, IsOwnerOrReadOnly
from .serializers import (
    TaskListSerializer, TaskDetailSerializer, CommentSerializer,
)


class TaskViewSet(viewsets.ModelViewSet):
    """
    CRUD для задач с permissions на уровне объекта.
    - list/create: любой аутентифицированный пользователь
    - retrieve: участник проекта задачи
    - update/destroy: assignee или owner проекта
    """

    filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
    filterset_class = TaskFilter
    search_fields = ["title", "description"]
    ordering_fields = ["created_at", "updated_at", "due_date", "priority", "status"]
    ordering = ["-created_at"]

    def get_permissions(self):
        if self.action in ("retrieve",):
            return [IsAuthenticated(), IsProjectMember()]
        if self.action in ("update", "partial_update", "destroy"):
            return [IsAuthenticated(), IsTaskAssigneeOrProjectOwner()]
        return [IsAuthenticated()]

    def get_queryset(self):
        return (
            Task.objects
            .filter(is_deleted=False)
            .select_related("project", "assignee", "project__owner")
            .prefetch_related("comments", "project__members")
        )

    def get_serializer_class(self):
        if self.action == "list":
            return TaskListSerializer
        return TaskDetailSerializer

    def perform_create(self, serializer):
        serializer.save()

    def perform_destroy(self, instance: Task):
        instance.is_deleted = True
        instance.save(update_fields=["is_deleted", "updated_at"])


class CommentViewSet(viewsets.ModelViewSet):
    """
    CRUD для комментариев.
    Удалять комментарий может только его автор.
    """

    serializer_class = CommentSerializer
    filter_backends = [DjangoFilterBackend, OrderingFilter]
    filterset_fields = ["task"]
    ordering_fields = ["created_at"]
    ordering = ["created_at"]

    def get_permissions(self):
        if self.action in ("update", "partial_update", "destroy"):
            return [IsAuthenticated(), IsOwnerOrReadOnly()]
        return [IsAuthenticated()]

    def get_queryset(self):
        return Comment.objects.select_related("author", "task", "task__project")

    def perform_create(self, serializer):
        """Автоматически назначаем автором текущего пользователя."""
        serializer.save(author=self.request.user)

🧠 Объяснение логики

Порядок вызова проверок в DRF

1. Аутентификация: DRF проверяет заголовок Authorization и устанавливает request.user
2. has_permission(): вызывается для каждого permission в списке — если хотя бы один вернул False, запрос отклоняется (403/401)
3. ViewSet выполняет action, получает объект из БД
4. check_object_permissions(): вызывается вручную (или автоматически в get_object()) — тригерит has_object_permission() для каждого permission

get_permissions() vs permission_classes

Атрибут permission_classes задаёт статический список permissions. Метод get_permissions() позволяет динамически менять список в зависимости от self.action. Используйте get_permissions(), когда для разных actions нужны разные permissions.

N+1 при проверке members

В IsProjectMember.has_object_permission() мы делаем project.members.filter(pk=...).exists(). Чтобы избежать лишних запросов, в get_queryset() добавлено prefetch_related("project__members"). Однако .filter().exists() всегда делает отдельный SQL даже с prefetch. Для высоконагруженных систем стоит кешировать список members в памяти.

✅ Проверка

1. Создание проекта (устанавливает owner)

# Получаем токен
ACCESS=$(curl -s -X POST http://127.0.0.1:8000/api/auth/token/ \
  -H "Content-Type: application/json" \
  -d '{"email": "user@example.com", "password": "Str0ngPass!"}' \
  | python -c "import sys,json; print(json.load(sys.stdin)['access'])")

# Создаём проект — owner устанавливается автоматически
curl -s -X POST http://127.0.0.1:8000/api/projects/ \
  -H "Authorization: Bearer $ACCESS" \
  -H "Content-Type: application/json" \
  -d '{"name": "Мой проект", "description": "Тест permissions"}' \
  | python -m json.tool

{
  "id": 1,
  "name": "Мой проект",
  "owner": {"id": 2, "email": "user@example.com", "username": "testuser"},
  "is_active": true
}

2. Попытка удалить чужой проект (должна вернуть 403)

# Логинимся другим пользователем
OTHER_ACCESS=$(curl -s -X POST http://127.0.0.1:8000/api/auth/token/ \
  -H "Content-Type: application/json" \
  -d '{"email": "other@example.com", "password": "OtherPass!"}' \
  | python -c "import sys,json; print(json.load(sys.stdin)['access'])")

# Пытаемся удалить проект другого пользователя
curl -s -X DELETE http://127.0.0.1:8000/api/projects/1/ \
  -H "Authorization: Bearer $OTHER_ACCESS" \
  -w "\nHTTP Status: %{http_code}\n"

{"detail": "Только владелец проекта может выполнять это действие."}

3. Создание комментария (author устанавливается автоматически)

curl -s -X POST http://127.0.0.1:8000/api/comments/ \
  -H "Authorization: Bearer $ACCESS" \
  -H "Content-Type: application/json" \
  -d '{"task": 1, "text": "Первый комментарий"}' \
  | python -m json.tool

Диагностика ошибок

• 403 при создании задачи — убедитесь, что в get_permissions() для create нет object-level проверки
• AttributeError: 'Task' has no 'owner' — в IsOwnerOrReadOnly.has_object_permission используется getattr(obj, "owner", None) or getattr(obj, "author", None) — проверьте оба атрибута
• project__members N+1 — добавьте prefetch_related("project__members") в get_queryset()

➡️ Что дальше

Права доступа настроены. Теперь нужно добавить реактивность: когда задача назначается исполнителю, он должен получить уведомление. Это делается через сигналы Django post_save.

• Готово: object-level permissions, автозаполнение owner/author при создании
• Шаг 10: сигналы post_save → email-уведомление assignee при назначении задачи