Как ограничить количественно выборку с помощью метода QuerySet?

Ограничение выборки с помощью QuerySet

Ограничение количества результатов важно для оптимизации производительности, правильной пагинации и экономии памяти. Рассмотрю основные методы.

Django ORM: QuerySet методы

1. Базовое ограничение: [:n]

from django.db import models
from myapp.models import User

# Получить первые 10 пользователей
users = User.objects.all()[:10]

# Эквивалентно SQL: LIMIT 10
print(users.query)  # SELECT * FROM users LIMIT 10

# Получить диапазон
users = User.objects.all()[5:15]  # Пропустить 5, взять 10
# SQL: SELECT * FROM users LIMIT 10 OFFSET 5

2. Методы first() и last()

# Получить первый объект или None
first_user = User.objects.filter(is_active=True).first()
# SELECT * FROM users WHERE is_active = true LIMIT 1

# Получить последний объект
last_user = User.objects.filter(is_active=True).last()

# Проверить наличие данных
if user := User.objects.filter(id=123).first():
    print(f"User found: {user.name}")

3. Методы exists() и count()

# Проверить наличие данных БЕЗ получения объектов
has_active = User.objects.filter(is_active=True).exists()
# SELECT 1 FROM users WHERE is_active = true LIMIT 1 - очень эффективно!

# Посчитать количество
total = User.objects.count()
# SELECT COUNT(*) FROM users

# Комбинирование
if User.objects.filter(email=email).exists():
    return "Email already registered"

SQLAlchemy

1. Базовое ограничение

from sqlalchemy import select
from sqlalchemy.orm import Session
from myapp.models import User

# limit() для ограничения
query = select(User).limit(10)
users = session.execute(query).scalars().all()
# SQL: SELECT * FROM users LIMIT 10

# offset() для пропуска
query = select(User).offset(5).limit(10)
users = session.execute(query).scalars().all()
# SQL: SELECT * FROM users LIMIT 10 OFFSET 5

2. first(), one(), one_or_none()

from sqlalchemy import select

# Получить первый результат
query = select(User).filter(User.is_active == True)
user = session.execute(query).scalars().first()

# Ровно один результат (или ошибка)
user = session.execute(query).scalars().one()

# Один или None
user = session.execute(query).scalars().one_or_none()

Пагинация

1. Классическая пагинация (offset/limit)

from math import ceil
from typing import Tuple

class Paginator:
    def __init__(self, query, page: int = 1, per_page: int = 20):
        self.query = query
        self.page = max(1, page)  # Минимум страница 1
        self.per_page = per_page
    
    def paginate(self) -> dict:
        # Получаем общее количество
        total = self.query.count()
        
        # Ограничиваем выборку
        offset = (self.page - 1) * self.per_page
        items = self.query[offset:offset + self.per_page]
        
        return {
            "items": items,
            "page": self.page,
            "per_page": self.per_page,
            "total": total,
            "pages": ceil(total / self.per_page)
        }

# Использование
users_query = User.objects.all()
paginator = Paginator(users_query, page=2, per_page=10)
result = paginator.paginate()
print(f"Page {result['page']} of {result['pages']}: {len(result['items'])} items")

2. Cursor-based пагинация (более эффективна)

from datetime import datetime
from typing import Optional
import base64

class CursorPaginator:
    """Эффективная пагинация для больших датасетов"""
    
    def __init__(self, query, cursor: Optional[str] = None, limit: int = 20):
        self.query = query
        self.cursor = cursor
        self.limit = limit
    
    def paginate(self) -> dict:
        # Если cursor есть, декодируем его
        if self.cursor:
            last_id = base64.b64decode(self.cursor).decode('utf-8')
            # Фильтруем по курсору
            items = self.query.filter(User.id > int(last_id)).limit(self.limit + 1)
        else:
            items = self.query.limit(self.limit + 1)
        
        items = list(items)
        has_next = len(items) > self.limit
        
        if has_next:
            items = items[:self.limit]
        
        # Генерируем новый cursor
        next_cursor = None
        if has_next and items:
            next_cursor = base64.b64encode(
                str(items[-1].id).encode('utf-8')
            ).decode('utf-8')
        
        return {
            "items": items,
            "next_cursor": next_cursor,
            "has_next": has_next
        }

# Использование в API
from fastapi import Query

@app.get("/users")
def list_users(cursor: Optional[str] = Query(None), limit: int = Query(20)):
    query = db.query(User).order_by(User.id)
    paginator = CursorPaginator(query, cursor, limit)
    result = paginator.paginate()
    
    return {
        "items": [user.to_dict() for user in result["items"]],
        "next_cursor": result["next_cursor"]
    }

Практические примеры

1. Получить последние 5 записей

# Django
last_5 = Post.objects.all().order_by('-created_at')[:5]

# SQLAlchemy
from sqlalchemy import desc
query = select(Post).order_by(desc(Post.created_at)).limit(5)
last_5 = session.execute(query).scalars().all()

2. Пропустить первые N и получить следующие M

# Django
users = User.objects.all()[10:20]  # Пропустить 10, получить 10

# SQLAlchemy
query = select(User).offset(10).limit(10)
users = session.execute(query).scalars().all()

3. Получить случайный элемент из большой таблицы

import random
from sqlalchemy import func

# Для малых таблиц
random_user = User.objects.order_by('?').first()  # Django
random_user = session.execute(
    select(User).order_by(func.random())
).scalars().first()  # SQLAlchemy

# Для больших таблиц (эффективнее)
count = User.objects.count()
random_offset = random.randint(0, count - 1)
random_user = User.objects.all()[random_offset]

4. Батчинг: обработка большой выборки по частям

def process_users_in_batches(batch_size: int = 1000):
    """Обработать всех пользователей без загрузки в память"""
    
    total = User.objects.count()
    
    for offset in range(0, total, batch_size):
        # Получаем батч
        batch = User.objects.all()[offset:offset + batch_size]
        
        # Обрабатываем
        for user in batch:
            process_user(user)
        
        print(f"Processed {offset + batch_size}/{total}")

# Альтернатива с iterator() - меньше памяти
for user in User.objects.iterator(chunk_size=1000):
    process_user(user)

5. Получить TOP N с GROUP BY

from django.db.models import Count

# Топ 5 пользователей по количеству комментариев
top_users = (
    User.objects.annotate(comment_count=Count('comments'))
    .order_by('-comment_count')[:5]
)

# SQL:
# SELECT users.*, COUNT(comments.id) as comment_count
# FROM users
# LEFT JOIN comments ON users.id = comments.user_id
# GROUP BY users.id
# ORDER BY comment_count DESC
# LIMIT 5

Оптимизация больших выборок

1. Используйте only() и defer() для выбора колонок

# Получить только нужные колонки
users = User.objects.only('id', 'name')[:1000]
# SQL: SELECT id, name FROM users LIMIT 1000

# Исключить тяжёлые колонки
users = User.objects.defer('bio', 'large_json_field')[:1000]

2. Используйте select_related() для JOIN

# Получить комментарии с авторами (N+1 problem!)
comments = Comment.objects.all()[:50]
for comment in comments:
    print(comment.author.name)  # Дополнительный запрос!

# Правильно с JOIN
comments = Comment.objects.select_related('author')[:50]
for comment in comments:
    print(comment.author.name)  # Данные уже в памяти

Чеклист для ограничения выборки

• Всегда используйте LIMIT для больших таблиц
• Для пагинации предпочитайте cursor-based вместо offset-based
• Логируйте запросы и проверяйте их в SQL
• Используйте only()/defer() для минимизации данных
• Используйте iterator() для батчинга больших выборок
• Проверяйте наличие N+1 problems
• Добавляйте индексы на колонки в WHERE и ORDER BY

Правильное ограничение выборки - ключ к производительности!