С помощью чего можно ограничить количество получаемых записей из БД?

Ограничение количества записей из БД

Это критичный навык для оптимизации производительности и управления нагрузкой на БД. Есть несколько подходов.

1. LIMIT в SQL запросе

Самый базовый и эффективный способ:

# SQLAlchemy ORM
from sqlalchemy import select
from sqlalchemy.orm import Session

# Ограничить 10 записей
users = session.query(User).limit(10).all()

# Или с SQLAlchemy 2.0+
users = session.execute(
    select(User).limit(10)
).scalars().all()

# Raw SQL
cursor.execute("SELECT * FROM users LIMIT 10")
results = cursor.fetchall()

Производительность: O(n), но БД останавливает сканирование после 10 записей.

2. OFFSET + LIMIT для пагинации

# Страница 2, по 10 записей на странице
page = 2
page_size = 10

users = session.query(User).offset((page - 1) * page_size).limit(page_size).all()

# SQLAlchemy 2.0+
users = session.execute(
    select(User)
    .offset((page - 1) * page_size)
    .limit(page_size)
).scalars().all()

# Raw SQL
cursor.execute(
    "SELECT * FROM users ORDER BY id LIMIT ? OFFSET ?",
    (page_size, (page - 1) * page_size)
)

Проблема: OFFSET медленнее на больших страницах (например, страница 1000).

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

Используем ID последней записи вместо OFFSET:

# Получить записи после ID 100
last_id = 100
page_size = 10

users = session.query(User).filter(
    User.id > last_id
).limit(page_size).all()

# Raw SQL
cursor.execute(
    "SELECT * FROM users WHERE id > ? ORDER BY id LIMIT ?",
    (last_id, page_size)
)

# Результат:
# ID: 101, 102, 103, ..., 110
# next_cursor = 110 (для следующего запроса)

Преимущества:

• O(1) вместо O(n) для OFFSET
• Не нужно считать количество записей
• Работает при добавлении новых записей

4. LIMIT для результатов подзапроса

# Получить top 10 постов по лайкам
top_posts = session.query(Post).order_by(
    Post.likes.desc()
).limit(10).all()

# SQLAlchemy
from sqlalchemy import func

top_comments = session.query(
    func.count(Comment.id).label("comment_count"),
    Post
).join(Comment).group_by(Post.id).order_by(
    func.count(Comment.id).desc()
).limit(5).all()

5. fetchone / fetchmany для больших результатов

Не загружаем всё в память сразу:

import psycopg2

conn = psycopg2.connect("dbname=myapp user=user")
cur = conn.cursor()

# Загрузить 1000000 записей, но обрабатывать по 1000
cur.execute("SELECT * FROM large_table")

while True:
    # Берём по 1000 записей
    rows = cur.fetchmany(1000)
    if not rows:
        break
    
    for row in rows:
        process(row)  # Обработать

cur.close()
conn.close()

Преимущества:

• Экономия памяти при работе с миллионами записей
• Можно обрабатывать данные потоком

6. Query.first() для получения одной записи

# Вместо limit(1).one() используй first()
first_user = session.query(User).filter(
    User.email == "user@example.com"
).first()  # Возвращает None если нет

# Эквивалент:
first_user = session.execute(
    select(User).where(User.email == "user@example.com").limit(1)
).scalar_one_or_none()  # None если нет записей

7. Слайсинг в ORM

# Django ORM
users = User.objects.all()[:10]  # Первые 10
users = User.objects.all()[10:20]  # С 10-й по 20-ю

# SQLAlchemy не поддерживает slice в query, но можно через limit+offset
users = session.query(User).limit(10)[0:10]  # Будет выполнено в памяти

8. Database-level pagination в FastAPI

from fastapi import FastAPI, Query
from sqlalchemy.orm import Session

app = FastAPI()

@app.get("/users")
def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),  # Максимум 100!
    db: Session = Depends(get_db)
):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

# Использование:
# GET /users?skip=0&limit=10
# GET /users?skip=10&limit=10 (страница 2)

Важно: Всегда ограничивайте максимальный limit!

9. COUNT с LIMIT для проверки наличия

# Вместо count() -> SELECT COUNT(*) который может быть медленным
# Проверим, есть ли хотя бы 1 запись

# ПЛОХО: SELECT COUNT(*) может долго считать миллионы
if session.query(User).filter(User.is_active == True).count() > 0:
    ...

# ХОРОШО: SELECT ... LIMIT 1 гораздо быстрее
if session.query(User).filter(User.is_active == True).limit(1).first():
    ...

# EXISTS субзапрос (самый быстрый)
from sqlalchemy import exists
if session.query(
    exists().where(User.is_active == True)
).scalar():
    ...

10. Stream результаты для больших объёмов

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("postgresql://...")

with Session(engine) as session:
    # Потоковые результаты (не загружаются все в памяти)
    for user in session.query(User).yield_per(1000):
        process(user)  # Обрабатываем по 1000 за раз

Практический пример: эффективная пагинация в API

from fastapi import FastAPI, Query, HTTPException
from sqlalchemy.orm import Session
from typing import Optional

app = FastAPI()

class PaginationParams:
    def __init__(
        self,
        limit: int = Query(20, ge=1, le=100),
        cursor: Optional[str] = Query(None)
    ):
        self.limit = limit
        self.cursor = cursor

@app.get("/users")
def list_users(
    params: PaginationParams = Depends(),
    db: Session = Depends(get_db)
):
    query = db.query(User)
    
    # Cursor-based pagination (эффективнее OFFSET)
    if params.cursor:
        try:
            last_id = int(params.cursor)
            query = query.filter(User.id > last_id)
        except ValueError:
            raise HTTPException(status_code=400, detail="Invalid cursor")
    
    # Ограничиваем на 1 больше, чтобы узнать, есть ли следующая страница
    users = query.order_by(User.id).limit(params.limit + 1).all()
    
    has_next = len(users) > params.limit
    if has_next:
        users = users[:params.limit]
    
    next_cursor = str(users[-1].id) if has_next and users else None
    
    return {
        "data": users,
        "next_cursor": next_cursor,
        "has_next": has_next
    }

Сравнение методов

Вывод

Ограничение записей критично для:

• Производительности (не загружаем всё в память)
• Безопасности (API не может быть перегружена)
• Пользовательского опыта (быстрая загрузка)

Рекомендации:

1. Всегда используйте LIMIT в запросах
2. Для пагинации предпочитайте cursor-based вместо OFFSET
3. Для больших объёмов используйте fetchmany() или yield_per()
4. Всегда ограничивайте максимальный limit в API (обычно 100)
5. Для проверки наличия используйте EXISTS вместо COUNT