Привет, Хабр!
Сегодня рассмотрим, как использовать F()‑экспрессии и Func()‑обёртки в Django для того, чтобы выполнять арифметику, условия и преобразования не в Python, а на стороне базы данных. Это важно, когда речь идёт о:
массовых обновлениях без
save(),атомарности,
производительности на высоких нагрузках,
и просто — умении думать в терминах SQL, не теряя удобства Django ORM.
F(): когда требуется значение из другого поля
Классическая ситуация — увеличить поле views:
article = Article.objects.get(id=1) article.views += 1 article.save()
Проблема: между get() и save() может произойти гонка данных. Если несколько воркеров обновляют значение одновременно — часть просмотров потеряется.
Атомарный способ:
from django.db.models import F Article.objects.filter(id=1).update(views=F('views') + 1)
Это превращается в SQL вида UPDATE article SET views = views + 1 WHERE id = 1. Никакой гонки. Никакой нагрузки на память.
Сравнение между полями
Фильтруем товары, у которых цена выше скидки:
Product.objects.filter(price__gt=F('discount_price'))
База сравнивает значения прямо внутри себя. Нет смысла загружать данные в Python, перебирать, сравнивать вручную. Работает с любыми типами: числа, даты, булевы, строки.
Вычисляемые поля через annotate()
Пусть нужно в выборке вывести стоимость товаров на складе:
from django.db.models import F, ExpressionWrapper, DecimalField Product.objects.annotate( stock_value=ExpressionWrapper( F('price') * F('stock'), output_field=DecimalField() ) )
Если используете DRF, stock_value будет доступен в сериализаторе как поле модели — без дополнительного кода.
Условные выражения: Case / When
Допустим, если stock == 0, хотим аннотировать строку "Нет в наличии", иначе "В наличии":
from django.db.models import Case, When, Value, CharField Product.objects.annotate( availability=Case( When(stock=0, then=Value("Нет в наличии")), default=Value("В наличии"), output_field=CharField() ) )
Полезно, когда хочется передать бизнес‑логику прямо в сериализатор, без дополнительной логики в Python.
Массовые обновления
В продакшене часто встречается кейс: если progress_percent >= 100, то status = 'completed'.
Вариант, который лучше забыть:
for task in Task.objects.all(): if task.progress_percent >= 100: task.status = 'completed' task.save()
Так делать нельзя: слишком много SQL‑запросов, лишняя нагрузка, долгие миграции.
Правильный подход:
Task.objects.filter(progress_percent__gte=100).update(status='completed')
SQL-функции через Func()
Хочется аннотировать имя пользователя в верхнем регистре? Без проблем:
from django.db.models import Func, F User.objects.annotate(name_upper=Func(F('name'), function='UPPER'))
А если требуется собрать full_name:
from django.db.models import Func, F, Value User.objects.annotate( full_name=Func( F('first_name'), Value(' '), F('last_name'), function='CONCAT' ) )
Всё выполняется на уровне SQL. Никаких .join() в Python, никакой логики в сериализаторах.
Автообновление updated_at через Now()
Когда нужно обновить updated_at, но не хочется вычислять timezone.now() в Python:
from django.db.models.functions import Now MyModel.objects.filter(condition=True).update(updated_at=Now())
Это NOW() в SQL. Удобно, особенно если работаете с select_for_update и хотите сохранить согласованность.
Собственные SQL-функции на Func()
Django позволяет кастомизировать Func, если нужен нестандартный SQL:
from django.db.models import Func, IntegerField class ExtractMonth(Func): function = 'EXTRACT' template = "%(function)s(MONTH FROM %(expressions)s)" output_field = IntegerField() # Использование: Invoice.objects.annotate(month=ExtractMonth('created_at'))
Работает на PostgreSQL, MySQL, Oracle — везде, где поддерживается EXTRACT.
Возможные проблемки
1. Без output_field в ExpressionWrapper — получите ошибку.
Django не умеет автоматически определять тип, если это не annotate(F() + 1). Поэтому:
ExpressionWrapper(F('a') * F('b'), output_field=DecimalField())
обязателен.
2. F() не сериализуется.
В DRF попытка передать F('views') + 1 в сериализатор вызовет ошибку. Это выражение, а не значение. Использовать можно только в .filter(), .update(), .annotate().
3. SQL‑функции зависят от базы.
Некоторые работают в PostgreSQL, но не в SQLite. Если проект использует SQLite в тестах — проверяйте поддержку.
Кейс
В CRM‑системе необходимо ежедневно пересчитывать KPI менеджеров по заказам:
score = orders_count 2 + late_orders -3 + F('bonus')status = "top"еслиscore > 50, иначе"regular"
Решение:
Manager.objects.annotate( score=ExpressionWrapper( F('orders_count') * 2 + F('late_orders') * -3 + F('bonus'), output_field=IntegerField() ) ).update( status=Case( When(score__gt=50, then=Value('top')), default=Value('regular'), output_field=CharField() ) )
Весь пересчёт — одним SQL‑запросом. Без фоновых задач. Без утечек.
Использование с select_for_update, Window, Subquery
F() внутри транзакции:
with transaction.atomic(): obj = Product.objects.select_for_update().get(id=1) obj.stock = F('stock') - 1 obj.save()
F() + Window‑функции:
from django.db.models import Window from django.db.models.functions import Rank Leaderboard.objects.annotate( rank=Window( expression=Rank(), order_by=F('score').desc() ) )
F() + Subquery:
from django.db.models import Subquery, OuterRef last_price = PriceHistory.objects.filter( product_id=OuterRef('pk') ).order_by('-date').values('price')[:1] Product.objects.annotate(last_price=Subquery(last_price))
🧾 Итого
Зачем нужны F() и Func():
Атомарность. Один SQL — одно действие. Без гонок.
Производительность. Работает там, где живут данные.
Выразительность. Пишется читаемо, читается понятно.
Безопасность. Исключает дублирование и сбои при массовых обновлениях.
Расширяемость. Поддерживает кастомные функции,
Case,Window,Subquery.
Не пропустите:
23 апреля, 20:00
«Асинхронное Django приложение: работа с асинхронными ORM и views»
Этот урок откроет новые горизонты для оптимизации асинхронных операций в Django. Подробности и регистрация на странице курса «Django‑разработчик».
