💻 Примеры: ORM-запросы и DRF views

Пример 1: Фильтрация проектов по текущему месяцу

from django.utils import timezone
from management_app.models.project import Project

# Получить проекты, созданные в текущем месяце
cur_month_projects = Project.objects.filter(
    date_of_creation__month=timezone.now().month
)

if cur_month_projects:
    for proj in cur_month_projects:
        print("=" * 50)
        print(proj.name)
        print(proj.date_of_creation)
else:
    print("EMPTY DATA")

Пример 2: ExtractWeekDay — файлы, загруженные в понедельник

from django.db.models.functions import ExtractWeekDay
from management_app.models.project import ProjectFile

monday_files = ProjectFile.objects.annotate(
    weekday=ExtractWeekDay('created_at')
).filter(
    weekday=2   # 2 = понедельник в ISO-стиле Django
)

if monday_files:
    for f in monday_files:
        print(f.file_name, f.file_path)
else:
    print("EMPTY DATA")

Пример 3: annotate + Avg + вывод результата

from django.db.models import Avg
from management_app.models import Project

# Среднее количество задач на каждом проекте
projects_with_avg = Project.objects.annotate(avg_tasks=Avg('tasks__id'))

for project in projects_with_avg:
    print(f"Project: {project.name}, Avg tasks: {project.avg_tasks}")

Пример 4: annotate + values_list + order_by

from django.db.models import Count
from django.contrib.auth.models import User

# Пользователи, отсортированные по убыванию количества задач
sorted_users = User.objects.annotate(
    count_of_tasks=Count('tasks__id')
).order_by('-count_of_tasks').values_list('username', 'count_of_tasks')

print(sorted_users.query)  # вывести SQL-запрос для отладки

for username, count in sorted_users:
    print(f"{username}  |  {count}")

Пример 5: Paginator — разбивка задач на страницы

from management_app.models import Task
from django.core.paginator import Paginator

tasks = Task.objects.all()
paginator = Paginator(tasks, 10)   # 10 задач на страницу

page_number = 1
page = paginator.get_page(page_number)

print(f"Страница {page.number} из {paginator.num_pages}")
for task in page:
    print("=" * 50)
    print(task.name)
    print(task.status)
    print(task.priority)
    print(task.assignee.username)

Пример 6: Полный сериализатор + view (DRF)

# management_app/serializers/projects.py
from rest_framework.serializers import ModelSerializer
from management_app.models import Project

class AllProjectsSerializer(ModelSerializer):
    class Meta:
        model = Project
        fields = ['id', 'name']


# management_app/views.py (или views/projects.py)
from rest_framework.request import Request
from rest_framework.decorators import api_view
from rest_framework import status
from django.http import JsonResponse
from management_app.serializers.projects import AllProjectsSerializer
from management_app.models.project import Project

@api_view(['GET'])
def get_all_projects(request: Request) -> JsonResponse:
    all_projects = Project.objects.all()
    if not all_projects.exists():
        return JsonResponse(
            [],
            status=status.HTTP_204_NO_CONTENT,
            safe=False
        )
    serialized_data = AllProjectsSerializer(all_projects, many=True)
    return JsonResponse(
        serialized_data.data,
        status=status.HTTP_200_OK,
        safe=False
    )

Пример 7: Вложенный сериализатор — TaskInfoSerializer

# management_app/serializers/tasks.py
from rest_framework import serializers
from management_app.models import Task
from management_app.serializers.tags import TagsSerializer

class TaskInfoSerializer(serializers.ModelSerializer):
    tags = TagsSerializer(many=True)  # теги — вложенный объект

    class Meta:
        model = Task
        fields = [
            'id', 'name', 'status', 'priority',
            'tags', 'project', 'created_date', 'due_date'
        ]


# Использование в view
@api_view(['GET'])
def get_task_by_id(request: Request, task_id: int) -> JsonResponse:
    try:
        task = Task.objects.get(id=task_id)
    except Task.DoesNotExist:
        return JsonResponse({}, status=status.HTTP_204_NO_CONTENT)
    data = TaskInfoSerializer(task)
    return JsonResponse(data.data, status=status.HTTP_200_OK)

Пример 8: CRUD для тегов — полный набор endpoints

# management_app/serializers/tags.py
from rest_framework import serializers
from management_app.models import Tag

class TagsSerializer(serializers.ModelSerializer):
    class Meta:
        model = Tag
        fields = '__all__'


# management_app/views.py — полный CRUD для Tag
@api_view(['GET'])
def get_tag_by_id(request: Request, tag_id: int) -> JsonResponse:
    try:
        tag = Tag.objects.get(id=tag_id)
    except Tag.DoesNotExist:
        return JsonResponse({}, status=status.HTTP_204_NO_CONTENT)
    return JsonResponse(TagsSerializer(tag).data, status=status.HTTP_200_OK)

@api_view(['PUT'])
def update_required_tag(request: Request, tag_id: int) -> JsonResponse:
    try:
        tag = Tag.objects.get(id=tag_id)
    except Tag.DoesNotExist:
        return JsonResponse({}, status=status.HTTP_204_NO_CONTENT)
    serializer = TagsSerializer(tag, request.data)
    if serializer.is_valid(raise_exception=True):
        serializer.save()
        return JsonResponse(serializer.data, status=status.HTTP_200_OK)

@api_view(['POST'])
def create_new_tag(request: Request) -> JsonResponse:
    new_tag = TagsSerializer(data=request.data)
    if new_tag.is_valid(raise_exception=True):
        new_tag.save()
        return JsonResponse(new_tag.data, status=status.HTTP_201_CREATED)
    return JsonResponse(new_tag.errors, status=status.HTTP_400_BAD_REQUEST)

@api_view(['DELETE'])
def delete_required_tag(request: Request, tag_id: int) -> JsonResponse:
    try:
        tag = Tag.objects.get(id=tag_id)
    except Tag.DoesNotExist:
        return JsonResponse({}, status=status.HTTP_204_NO_CONTENT)
    tag.delete()
    return JsonResponse({'message': 'Tag deleted successfully'}, status=status.HTTP_200_OK)

← К оглавлению урока