✅ Решения: Agile Projects ч.2
⚡ Решения — ключевые моменты
- Задание 1:
partial=TrueвCreateProjectSerializer(instance=..., data=..., partial=True) - Задание 2: валидаторы — на уровне класса (не внутри Meta),
paginator = StandardResultsSetPaginationкак атрибут класса - Задание 3:
TaskDetailSerializerсexclude = ('updated_at', 'deleted_at'),update()черезsetattr - Задание 4:
file.chunks()в save_file, контекст сериализатора:context={'raw_file': ..., 'project': ...} - Задание 5:
os.remove(os.path.realpath(file_path))для физического удаления - Задание 6:
AUTH_USER_MODEL = 'users.User'в settings.py до миграций
Решение задания 1: ProjectDetailAPIView
1. Сериализатор ProjectDetailSerializer
# apps/projects/serializers/project_serializers.py
class ProjectDetailSerializer(serializers.ModelSerializer):
class Meta:
model = Project
fields = ('id', 'name', 'description', 'count_of_files')
2. Класс-отображение ProjectDetailAPIView
# apps/projects/views/project_views.py
from rest_framework.views import APIView
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework import status
from rest_framework.generics import get_object_or_404
class ProjectDetailAPIView(APIView):
def get_object(self, pk: int):
return get_object_or_404(Project, pk=pk)
def get(self, request: Request, pk: int) -> Response:
project = self.get_object(pk=pk)
serializer = ProjectDetailSerializer(project)
return Response(serializer.data, status=status.HTTP_200_OK)
def put(self, request: Request, pk: int) -> Response:
project = self.get_object(pk=pk)
serializer = CreateProjectSerializer(
instance=project,
data=request.data,
partial=True # <-- частичное обновление полей
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.validated_data, status=status.HTTP_200_OK)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
def delete(self, request: Request, pk: int) -> Response:
project = self.get_object(pk=pk)
project.delete()
return Response(
data={"message": "Project was deleted successfully"},
status=status.HTTP_200_OK
)
3. Регистрация эндпоинта
# apps/projects/urls.py
urlpatterns = [
path('', ProjectsListAPIView.as_view()),
path('<int:pk>/', ProjectDetailAPIView.as_view()), # NEW
]
4. Запустить сервер, проверить GET, PUT, DELETE.
5. git add . && git commit -m "feat: task1 — ProjectDetailAPIView" && git push origin <branch>
Решение задания 2: TasksListAPIView с пагинацией
1. Сериализаторы AllTasksSerializer и CreateTaskSerializer
# apps/tasks/serializers/task_serializers.py
from typing import Any
from rest_framework import serializers
from django.utils import timezone
from apps.projects.models import Project
from apps.tasks.models import Task, Tag
from apps.tasks.choices.priority import Priority
class AllTasksSerializer(serializers.ModelSerializer):
project = serializers.SlugRelatedField(read_only=True, slug_field='name')
assignee = serializers.SlugRelatedField(read_only=True, slug_field='email')
class Meta:
model = Task
fields = ('id', 'name', 'status', 'priority', 'project', 'assignee', 'deadline')
class CreateTaskSerializer(serializers.ModelSerializer):
project = serializers.SlugRelatedField(
slug_field='name',
queryset=Project.objects.all(),
)
class Meta:
model = Task
fields = ('name', 'description', 'priority', 'project', 'tags', 'deadline')
# ✅ validate_* — НА УРОВНЕ КЛАССА, не внутри Meta
def validate_name(self, value: str) -> str:
if len(value) < 10:
raise serializers.ValidationError(
"The name of the task couldn't be less than 10 characters"
)
return value
def validate_description(self, value: str) -> str:
if len(value) < 50:
raise serializers.ValidationError(
"The description of the task couldn't be less than 50 characters"
)
return value
def validate_priority(self, value: int) -> int:
if value not in [val[0] for val in Priority.choices()]:
raise serializers.ValidationError(
"The priority of the task couldn't be one of the available options"
)
return value
def validate_project(self, value) -> ...:
# SlugRelatedField уже ищет по имени — если не найдено, DRF выдаст ошибку
# Дополнительная проверка для кастомного сообщения:
if not Project.objects.filter(name=value.name if hasattr(value, 'name') else value).exists():
raise serializers.ValidationError(
"The project with this name couldn't be found in the database"
)
return value
def validate_tags(self, value) -> ...:
if value and not Tag.objects.filter(pk__in=[t.pk for t in value]).exists():
raise serializers.ValidationError("The tags couldn't be found in the database")
return value
def validate_deadline(self, value) -> ...:
if timezone.is_naive(value):
value = timezone.make_aware(value, timezone.get_current_timezone())
if value < timezone.now():
raise serializers.ValidationError(
"The deadline of the task couldn't be in the past"
)
return value
def create(self, validated_data: dict[str, Any]) -> Task:
tags = validated_data.pop('tags', [])
task = Task.objects.create(**validated_data)
for tag in tags:
task.tags.add(tag)
task.save()
return task
2. Класс-отображение TasksListAPIView
# apps/tasks/views/task_views.py
from django.db.models import QuerySet
from rest_framework.response import Response
from rest_framework.request import Request
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.pagination import PageNumberPagination
from apps.tasks.models import Task
from apps.tasks.serializers.task_serializers import AllTasksSerializer, CreateTaskSerializer
class StandardResultsSetPagination(PageNumberPagination):
page_size = 5
page_size_query_param = 'page_size'
max_page_size = 15
class TasksListAPIView(APIView):
paginator = StandardResultsSetPagination
def get_objects(self) -> QuerySet:
project_name = self.request.query_params.get('project_name')
assignee_email = self.request.query_params.get('assignee_email')
if project_name:
return Task.objects.filter(project__name=project_name)
elif assignee_email:
return Task.objects.filter(assignee__email=assignee_email)
return Task.objects.all()
def get(self, request: Request, *args, **kwargs) -> Response:
tasks = self.get_objects()
if not tasks.exists():
return Response(data=[], status=status.HTTP_204_NO_CONTENT)
paginator = self.paginator()
page = paginator.paginate_queryset(tasks, request, view=self)
if page is not None:
serializer = AllTasksSerializer(page, many=True)
return paginator.get_paginated_response(serializer.data)
serializer = AllTasksSerializer(tasks, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def post(self, request: Request, *args, **kwargs) -> Response:
serializer = CreateTaskSerializer(data=request.data) # исправлена опечатка из лекции
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
3. Регистрация эндпоинта
# apps/tasks/urls.py
from apps.tasks.views.task_views import TasksListAPIView
urlpatterns = [
path('', TasksListAPIView.as_view()), # NEW
path('tags/', TagListAPIView.as_view()),
path('tags/<int:pk>/', TagDetailAPIView.as_view()),
]
Решение задания 3: TaskDetailAPIView
1. Сериализатор ProjectShortInfoSerializer
# apps/projects/serializers/project_serializers.py
class ProjectShortInfoSerializer(serializers.ModelSerializer):
class Meta:
model = Project
fields = ('id', 'name')
2. Сериализатор TaskDetailSerializer
# apps/tasks/serializers/task_serializers.py
class TaskDetailSerializer(serializers.ModelSerializer):
project = ProjectShortInfoSerializer() # вложенный сериализатор
tags = TagSerializer(many=True, read_only=True)
class Meta:
model = Task
exclude = ('updated_at', 'deleted_at')
3. Переименование и метод update в CreateUpdateTaskSerializer
class CreateUpdateTaskSerializer(serializers.ModelSerializer):
project = serializers.SlugRelatedField(
slug_field='name',
queryset=Project.objects.all(),
)
# ... валидаторы и create() остаются ...
def update(self, instance: Task, validated_data: dict[str, Any]) -> Task:
tags = validated_data.pop('tags', [])
for attr, value in validated_data.items():
setattr(instance, attr, value)
if tags:
for tag in tags:
instance.tags.add(tag)
instance.save()
return instance
4. Класс-отображение TaskDetailAPIView
class TaskDetailAPIView(APIView):
def get_object(self):
return get_object_or_404(Task, pk=self.kwargs['pk'])
def get(self, request: Request, *args, **kwargs) -> Response:
task = self.get_object()
serializer = TaskDetailSerializer(task)
return Response(serializer.data, status=status.HTTP_200_OK)
def put(self, request: Request, *args, **kwargs) -> Response:
task = self.get_object()
serializer = CreateUpdateTaskSerializer(
instance=task, data=request.data, partial=True
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(serializer.data, status=status.HTTP_200_OK)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
def delete(self, request: Request, *args, **kwargs) -> Response:
task = self.get_object()
task.delete()
return Response(
data={"message": "The task has been deleted."},
status=status.HTTP_204_NO_CONTENT
)
Решение задания 4: Загрузка файлов
1. upload_file_helpers.py
# apps/projects/utils/upload_file_helpers.py
import os
from pathlib import Path
ALLOWED_EXTENSIONS = ['.csv', '.doc', '.pdf', '.xlsx']
def check_extension(filename: str) -> bool:
extension = Path(filename).suffix
return extension in ALLOWED_EXTENSIONS
def check_file_size(file, required_size: int = 2) -> bool:
file_size = file.size / (1024 * 1024)
return file_size <= required_size
def create_file_path(project_name: str, file_name: str) -> str:
# rsplit безопасен для имён с несколькими точками
stem, ext = file_name.rsplit('.', 1)
file_path = "documents/{}/{}.{}".format(
project_name.replace(' ', '_'),
stem.replace(' ', '_'),
ext
)
return file_path
def save_file(file_path: str, file_content) -> str:
os.makedirs(os.path.dirname(Path(file_path)), exist_ok=True)
with open(file_path, 'wb') as f:
for chunk in file_content.chunks(): # запись по частям
f.write(chunk)
return file_path
2. project_file_serializers.py
# apps/projects/serializers/project_file_serializers.py
from rest_framework import serializers
from apps.projects.models import ProjectFile
from apps.projects.utils.upload_file_helpers import (
check_extension, create_file_path, check_file_size, save_file
)
class AllProjectFilesSerializer(serializers.ModelSerializer):
class Meta:
model = ProjectFile
fields = ('id', 'file_name', 'project')
class CreateProjectFileSerializer(serializers.ModelSerializer):
class Meta:
model = ProjectFile
fields = ('file_name',)
def validate_file_name(self, value: str) -> str:
if not value.isascii():
raise serializers.ValidationError("Please, provide a valid file name.")
if not check_extension(value):
raise serializers.ValidationError(
"Valid file extensions: ['.csv', '.doc', '.pdf', '.xlsx']"
)
return value
def create(self, validated_data):
project = self.context.get('project')
raw_file = self.context.get('raw_file')
file_path = create_file_path(
project_name=project.name,
file_name=validated_data['file_name']
)
if check_file_size(file=raw_file):
save_file(file_path=file_path, file_content=raw_file)
validated_data['file_path'] = file_path
project_file = ProjectFile.objects.create(**validated_data)
project_file.project.add(project)
return project_file
else:
raise serializers.ValidationError("File size is too large (2 MB as maximum).")
3. ProjectFileListGenericView
# apps/projects/views/project_file_views.py
from rest_framework.generics import ListCreateAPIView, get_object_or_404
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework import status
from apps.projects.models import ProjectFile, Project
class ProjectFileListGenericView(ListCreateAPIView):
def get_serializer_class(self, *args, **kwargs):
if self.request.method == 'GET':
return AllProjectFilesSerializer
return CreateProjectFileSerializer
def get_queryset(self):
project_name = self.request.query_params.get('project')
if project_name:
return ProjectFile.objects.filter(project__name=project_name)
return ProjectFile.objects.all()
def list(self, request: Request, *args, **kwargs) -> Response:
project_files = self.get_queryset()
if not project_files.exists():
return Response(data=[], status=status.HTTP_204_NO_CONTENT)
serializer = self.get_serializer(project_files, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def create(self, request: Request, *args, **kwargs) -> Response:
file_content = request.FILES["file"]
project_id = request.data["project_id"]
request.data['file_name'] = file_content.name if file_content else None
project = get_object_or_404(Project, pk=project_id)
serializer = self.get_serializer(
data=request.data,
context={"raw_file": file_content, "project": project}
)
if serializer.is_valid(raise_exception=True):
serializer.save()
return Response(data={"message": "File upload successfully"}, status=status.HTTP_200_OK)
return Response(data=serializer.errors, status=status.HTTP_400_BAD_REQUEST)
4. URL-регистрация
# apps/projects/urls.py
urlpatterns = [
path('', ProjectsListAPIView.as_view()),
path('<int:pk>/', ProjectDetailAPIView.as_view()),
path('files/', ProjectFileListGenericView.as_view()), # NEW
]
Решение задания 5: ProjectFileDetailGenericView
1. ProjectFileDetailSerializer
class ProjectFileDetailSerializer(serializers.ModelSerializer):
project = ProjectShortInfoSerializer(many=True)
class Meta:
model = ProjectFile
exclude = ('file_path',)
2. ProjectFileDetailGenericView
from rest_framework.generics import RetrieveDestroyAPIView
from apps.projects.utils.upload_file_helpers import delete_file
class ProjectFileDetailGenericView(RetrieveDestroyAPIView):
serializer_class = ProjectFileDetailSerializer
def get_object(self):
return get_object_or_404(ProjectFile, pk=self.kwargs['pk'])
def retrieve(self, request: Request, *args, **kwargs) -> Response:
task = self.get_object()
serializer = self.get_serializer(task)
return Response(data=serializer.data, status=status.HTTP_200_OK)
def destroy(self, request: Request, *args, **kwargs) -> Response:
task = self.get_object()
try:
delete_file(file_path=task.file_path.path)
except Exception as e:
return Response(data={"message": str(e)}, status=status.HTTP_400_BAD_REQUEST)
else:
task.delete()
return Response(data={"message": "File deleted successfully"}, status=status.HTTP_200_OK)
3. Функция delete_file
# apps/projects/utils/upload_file_helpers.py
def delete_file(file_path: str) -> None:
os.remove(os.path.realpath(file_path))
4. URL-регистрация
urlpatterns = [
path('', ProjectListAPIView.as_view()),
path('<int:pk>/', ProjectDetailAPIView.as_view()),
path('files/', ProjectFileListGenericView.as_view()),
path('files/<int:pk>/', ProjectFileDetailGenericView.as_view()), # NEW
]
Решение задания 6: Кастомный User
1. Создание приложения
python manage.py startapp users
2. Модель User (apps/users/models.py)
from django.contrib.auth.base_user import AbstractBaseUser
from django.contrib.auth.models import PermissionsMixin, UserManager
from django.core.validators import MinLengthValidator
from django.db import models
from django.utils.translation import gettext_lazy as _
from apps.projects.models import Project
from apps.users.choices.positions import UserPositions
class User(AbstractBaseUser, PermissionsMixin):
username = models.CharField(
_("username"), max_length=50, unique=True,
error_messages={"unique": _("A user with that username already exists.")}
)
first_name = models.CharField(_("first name"), max_length=40, validators=[MinLengthValidator(2)])
last_name = models.CharField(_("last name"), max_length=40, validators=[MinLengthValidator(2)])
email = models.EmailField(_("email address"), max_length=150, unique=True)
phone = models.CharField(max_length=75, null=True, blank=True)
is_staff = models.BooleanField(default=False)
is_active = models.BooleanField(default=True)
date_joined = models.DateTimeField(name="registered", auto_now_add=True)
last_login = models.DateTimeField(null=True, blank=True)
updated_at = models.DateTimeField(auto_now=True)
deleted_at = models.DateTimeField(null=True, blank=True)
deleted = models.BooleanField(default=False) # исправлено: False, не True
position = models.CharField(
max_length=15, choices=UserPositions.choices, default=UserPositions.PROGRAMMER
)
project = models.ForeignKey(
Project, on_delete=models.CASCADE, related_name="users", null=True, blank=True
)
USERNAME_FIELD = "email"
REQUIRED_FIELDS = ["username", "first_name", "last_name", "position"]
objects = UserManager()
def __str__(self):
return f"{self.last_name} {self.first_name}"
3. Choices позиций (apps/users/choices/positions.py)
from django.db import models
class UserPositions(models.TextChoices):
CEO = "Ceo", "CEO"
CTO = "Cto", "CTO"
DESIGNER = "Designer", "Designer"
PROGRAMMER = "Programmer", "Programmer"
PRODUCT_OWNER = "Product_owner", "Product Owner"
PROJECT_OWNER = "Project_owner", "Project Owner"
PROJECT_MANAGER = "Project_manager", "Project Manager"
QA = "Qa", "QA"
4. Settings
# settings.py — ДОБАВИТЬ до первых миграций
AUTH_USER_MODEL = 'users.User'
INSTALLED_APPS = [
...
'apps.users.apps.UsersConfig',
]
5. Обновить ForeignKey в модели Task
# apps/tasks/models/task_model.py
from django.conf import settings
class Task(models.Model):
...
assignee = models.ForeignKey(
settings.AUTH_USER_MODEL, # было: 'auth.User' или User
on_delete=models.SET_NULL,
null=True, blank=True,
related_name='tasks'
)
6. Пересоздание БД и миграции
# Удалить db.sqlite3 и все файлы migrations/ (кроме __init__.py)
# Затем:
python manage.py makemigrations
python manage.py migrate
python manage.py createsuperuser
Проверка: После
createsuperuser Django попросит email (не username) — это подтверждает, что USERNAME_FIELD = "email" работает корректно.