select_related, prefetch_related, F(), Q(), custom managers, raw SQL
Этот туториал охватывает продвинутые техники работы с Django ORM: оптимизация запросов, кастомные менеджеры, raw SQL, транзакции.
💡 Правило: Прежде чем писать raw SQL, спросите себя: "Можно ли сделать это через ORM?" Если да — делайте через ORM. Если нет — документируйте почему.
# ❌ N+1 проблема
posts = Post.objects.all()
for post in posts:
print(post.author.name) # N+1 запросов
# ✅ select_related для ForeignKey/OneToOne
posts = Post.objects.select_related('author')
for post in posts:
print(post.author.name) # 1 запрос с JOIN
# Глубокая связь
posts = Post.objects.select_related('author__profile')
# Несколько связей
posts = Post.objects.select_related('author', 'category')Когда использовать:
# ❌ N+1 проблема
posts = Post.objects.all()
for post in posts:
for tag in post.tags.all(): # N+1 запросов
print(tag.name)
# ✅ prefetch_related для ManyToMany/reverse FK
posts = Post.objects.prefetch_related('tags')
for post in posts:
for tag in post.tags.all(): # 2 запроса всего
print(tag.name)
# С фильтрацией связанных объектов
from django.db.models import Prefetch
posts = Post.objects.prefetch_related(
Prefetch(
'comments',
queryset=Comment.objects.filter(is_approved=True).select_related('author'),
to_attr='approved_comments' # Сохранить в атрибут
)
)Когда использовать:
| Характеристика | select_related | prefetch_related |
|---|---|---|
| Тип связи | ForeignKey, OneToOne | ManyToMany, reverse FK |
| Метод | SQL JOIN | Отдельные запросы |
| Объединение | На уровне БД | На уровне Python |
| Глубина | Неограниченная | Неограниченная |
| Производительность | Лучше для 1:1 | Лучше для 1:N, M:N |
from django.db.models import F
# ❌ Race condition
post = Post.objects.get(pk=1)
post.views += 1
post.save() # Может потерять обновления
# ✅ Атомарное увеличение
Post.objects.filter(pk=1).update(views=F('views') + 1)
# Сложные выражения
Post.objects.filter(pk=1).update(
rating=F('likes') * 2 + F('views') * 0.1
)# Найти посты где просмотров больше чем лайков
Post.objects.filter(views__gt=F('likes'))
# Найти товары где цена больше скидки
Product.objects.filter(price__gt=F('discount'))
# Аннотация с F()
from django.db.models import Value
Post.objects.annotate(
engagement=F('likes') + F('comments') * 2
).filter(engagement__gt=100)# ❌ Проблема гонки
def increment_views(post_id):
post = Post.objects.get(pk=post_id)
post.views += 1
post.save()
# ✅ С F() выражением
def increment_views(post_id):
Post.objects.filter(pk=post_id).update(views=F('views') + 1)
# ✅ С транзакцией и select_for_update
from django.db import transaction
def increment_views_safe(post_id):
with transaction.atomic():
post = Post.objects.select_for_update().get(pk=post_id)
post.views += 1
post.save()from django.db.models import Q
# OR условие
Post.objects.filter(
Q(author__username='john') | Q(author__username='jane')
)
# NOT условие
Post.objects.filter(
~Q(status='draft')
)
# Комбинирование
Post.objects.filter(
Q(is_published=True) &
(Q(author__username='john') | Q(author__username='jane'))
)
# Упрощённая запись (AND подразумевается)
Post.objects.filter(
is_published=True,
Q(author__username='john') | Q(author__username='jane')
)def search_posts(filters):
queryset = Post.objects.filter(is_published=True)
q_objects = Q()
if filters.get('author'):
q_objects |= Q(author__username=filters['author'])
if filters.get('category'):
q_objects |= Q(category__slug=filters['category'])
if filters.get('tags'):
for tag in filters['tags'].split(','):
q_objects |= Q(tags__name__icontains=tag.strip())
if q_objects:
queryset = queryset.filter(q_objects)
return queryset.distinct()class PublishedManager(models.Manager):
"""Менеджер для опубликованных постов."""
def get_queryset(self):
return super().get_queryset().filter(
is_published=True,
status='published'
)
def with_comments(self):
return self.get_queryset().annotate(
comment_count=Count('comments')
).filter(comment_count__gt=0)
def popular(self, min_views=100):
return self.get_queryset().filter(views__gte=min_views)
class Post(models.Model):
title = models.CharField(max_length=200)
is_published = models.BooleanField(default=False)
status = models.CharField(max_length=20, default='draft')
# Менеджеры
objects = models.Manager() # По умолчанию (все объекты)
published = PublishedManager() # Только опубликованные
class Meta:
# По умолчанию использовать published
base_manager_name = 'published'Использование:
Post.objects.all() # Все посты
Post.published.all() # Только опубликованные
Post.published.with_comments().popular() # Цепочка методовclass PostManager(models.Manager):
def create_post(self, title, author, content='', **kwargs):
"""Создать пост с валидацией."""
if not title:
raise ValueError('Title is required')
post = self.model(
title=title,
author=author,
content=content,
**kwargs
)
post.save(using=self._db)
return post
def create_published(self, title, author, **kwargs):
"""Создать опубликованный пост."""
kwargs['is_published'] = True
kwargs['status'] = 'published'
kwargs['published_at'] = timezone.now()
return self.create_post(title, author, **kwargs)
class Post(models.Model):
objects = PostManager()# Простой raw запрос
posts = Post.objects.raw('SELECT * FROM blog_post WHERE is_published = TRUE')
# С параметрами (безопасно!)
author_id = 5
posts = Post.objects.raw(
'SELECT * FROM blog_post WHERE author_id = %s',
[author_id]
)
# С именованными параметрами
posts = Post.objects.raw(
'SELECT * FROM blog_post WHERE author_id = %(author_id)s',
{'author_id': author_id}
)from django.db import connection
# UPDATE
with connection.cursor() as cursor:
cursor.execute(
'UPDATE blog_post SET views = views + 1 WHERE id = %s',
[post_id]
)
# SELECT с fetch
with connection.cursor() as cursor:
cursor.execute(
'SELECT title, views FROM blog_post WHERE author_id = %s',
[author_id]
)
rows = cursor.fetchall() # Все строки
# row = cursor.fetchone() # Одна строка# ❌ УЯЗВИМОСТЬ! Никогда не делайте так!
user_input = "1; DROP TABLE blog_post; --"
Post.objects.raw(f"SELECT * FROM blog_post WHERE id = {user_input}")
# ✅ Безопасно: параметризованные запросы
Post.objects.raw(
"SELECT * FROM blog_post WHERE id = %s",
[user_input]
)from django.db import transaction
# Как декоратор
@transaction.atomic
def transfer_money(from_account, to_account, amount):
from_account.balance -= amount
from_account.save()
to_account.balance += amount
to_account.save()
# Если ошибка — всё откатится
# Как контекстный менеджер
def transfer_money(from_account, to_account, amount):
with transaction.atomic():
from_account.balance -= amount
from_account.save()
to_account.balance += amount
to_account.save()
# Вложенные транзакции (savepoints)
@transaction.atomic
def outer():
with transaction.atomic(): # savepoint
# ...
raise Exception # Откатится до savepoint, не до начала outer()from django.db import transaction
@transaction.atomic
def withdraw(account_id, amount):
# Блокировка строки до конца транзакции
account = Account.objects.select_for_update().get(pk=account_id)
if account.balance >= amount:
account.balance -= amount
account.save()
return True
return Falsefrom django.db.models import OuterRef, Subquery
# Последний пост каждого автора
latest_post = Post.objects.filter(
author=OuterRef('pk')
).order_by('-created_at').values('id')[:1]
authors = Author.objects.annotate(
latest_post_id=Subquery(latest_post)
)
# Количество комментариев к последнему посту
latest_post_comments = Post.objects.filter(
author=OuterRef('pk')
).order_by('-created_at').values('id')[:1]
authors = Author.objects.annotate(
latest_post_comment_count=Subquery(
Comment.objects.filter(
post_id=Subquery(latest_post_comments)
).values('id').annotate(
count=Count('id')
).values('count')
)
)from django.db.models import Exists
# Авторы у которых есть опубликованные посты
has_posts = Post.objects.filter(
author=OuterRef('pk'),
is_published=True
)
authors = Author.objects.annotate(
has_published_posts=Exists(has_posts)
).filter(has_published_posts=True)# blog/analytics.py
from django.db.models import (
Count, Sum, Avg, F, Q, OuterRef, Subquery, Exists
)
from django.db import connection
from blog.models import Post, Author, Comment
class BlogAnalytics:
"""Класс для аналитики блога."""
@staticmethod
def top_authors_by_views(limit=10):
"""Топ авторов по просмотрам."""
return Author.objects.annotate(
total_views=Sum('posts__views'),
post_count=Count('posts'),
avg_views=Avg('posts__views')
).order_by('-total_views')[:limit]
@staticmethod
def popular_posts_with_engagement(min_views=100):
"""Популярные посты с вовлечённостью."""
return Post.objects.filter(
is_published=True,
views__gte=min_views
).select_related('author').prefetch_related('tags').annotate(
comment_count=Count('comments'),
like_count=Count('likes'),
engagement_score=F('views') + F('comments') * 10 + F('likes') * 5
).order_by('-engagement_score')
@staticmethod
def authors_with_recent_activity(days=7):
"""Авторы с активностью за последние N дней."""
from django.utils import timezone
from datetime import timedelta
recent_date = timezone.now() - timedelta(days=days)
has_recent_post = Post.objects.filter(
author=OuterRef('pk'),
created_at__gte=recent_date
)
return Author.objects.annotate(
has_recent=Exists(has_recent_post),
recent_post_count=Count(
'posts',
filter=Q(posts__created_at__gte=recent_date)
)
).filter(has_recent=True)
@staticmethod
def post_performance_metrics(post_id):
"""Метрики производительности поста."""
with connection.cursor() as cursor:
cursor.execute("""
SELECT
p.id,
p.title,
p.views,
COUNT(DISTINCT c.id) as comment_count,
COUNT(DISTINCT l.id) as like_count,
AVG(CASE WHEN c.is_approved THEN 1 ELSE 0 END) as approval_rate
FROM blog_post p
LEFT JOIN blog_comment c ON p.id = c.post_id
LEFT JOIN blog_like l ON p.id = l.post_id
WHERE p.id = %s
GROUP BY p.id, p.title, p.views
""", [post_id])
return cursor.fetchone()
@staticmethod
def trending_tags(days=30):
"""Трендовые теги."""
from django.utils import timezone
from datetime import timedelta
recent_date = timezone.now() - timedelta(days=days)
return Post.objects.filter(
created_at__gte=recent_date,
is_published=True
).prefetch_related('tags').annotate(
total_engagement=F('views') + Count('comments') * 10
).values('tags__name').annotate(
total_views=Sum('views'),
total_posts=Count('id'),
avg_engagement=Avg('total_engagement')
).order_by('-total_views')Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.