Что делать, если запрос начинает выполняться значительно дольше?

Отладка медленного запроса: Пошаговое руководство

Когда запрос "вдруг" начинает выполняться медленнее — это обычно означает проблему с данными, квери или ресурсами. Вот методический подход к отладке.

Шаг 1: Получить точные метрики

Вариант 1: Логирование времени

import time
from datetime import datetime

# FastAPI
@app.get("/api/users/{user_id}")
async def get_user(user_id: int):
    start_time = time.time()
    
    # Логируем каждый этап
    t1 = time.time()
    user = await fetch_user_from_db(user_id)
    t2 = time.time()
    print(f"DB query: {t2-t1:.2f}s")
    
    t3 = time.time()
    comments = await fetch_comments(user_id)
    t4 = time.time()
    print(f"Fetch comments: {t4-t3:.2f}s")
    
    total = time.time() - start_time
    print(f"Total time: {total:.2f}s")
    
    return {"user": user, "comments": comments}

Вариант 2: Django Debug Toolbar

# settings.py
DEBUG = True
INSTALLED_APPS = [
    # ...
    "debug_toolbar",
]

MIDDLEWARE = [
    # ...
    "debug_toolbar.middleware.DebugToolbarMiddleware",
]

# urls.py
if DEBUG:
    urlpatterns += [path("__debug__/", include("debug_toolbar.urls"))]

Открыв страницу в браузере, видишь все SQL квери и их время.

Вариант 3: Профилирование с cProfile

import cProfile
import pstats
from io import StringIO

def slow_function():
    # Долгая операция
    total = 0
    for i in range(1000000):
        total += i
    return total

# Профилируем
pr = cProfile.Profile()
pr.enable()

slow_function()

pr.disable()
s = StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
ps.print_stats(10)  # Топ 10 функций
print(s.getvalue())

Вывод показывает, какие функции занимают больше всего времени.

Шаг 2: Определить узкое место

Контрольный список

❓ Это проблема Database?
  → SELECT * выполняется 30 секунд вместо 0.1s
  → N+1 queries (1000 запросов вместо 1)
  → Отсутствуют индексы
  → Блокировка (lock) на таблице

❓ Это проблема вычислений?
  → Обработка больших массивов данных (100k+ элементов)
  → Сложный алгоритм O(n²) вместо O(n log n)
  → Неправильная кэширование

❓ Это проблема Network?
  → Запрос к внешнему API (к Telegram, к платёжной системе)
  → Медленное соединение (500ms вместо 10ms)
  → Таймаут и retry

❓ Это проблема памяти?
  → Memory leak (используется всё больше памяти)
  → Swap (диск используется вместо памяти)
  → Garbage collection паузы

Шаг 3: Отладка по типам проблем

Проблема: N+1 Queries

# БЕЗ оптимизации: 1000 пользователей = 1001 запрос
users = User.objects.all()
for user in users:
    print(user.profile.bio)  # Отдельный запрос для каждого!

# С оптимизацией: 2 запроса
users = User.objects.select_related("profile").all()
for user in users:
    print(user.profile.bio)  # Уже загружено

# Проверка в Django Debug Toolbar:
# БЕЗ: 1001 queries
# С: 2 queries

Проблема: Отсутствие индекса

-- БЕЗ индекса: 30 секунд на 1 млн записей
SELECT * FROM orders WHERE user_id = 123;

-- С индексом: 0.1 секунда
CREATE INDEX idx_orders_user_id ON orders(user_id);
SELECT * FROM orders WHERE user_id = 123;

Проверка в PostgreSQL:

-- Какие индексы есть?
\d orders

-- Статистика запроса
EXPLAIN ANALYZE
SELECT * FROM orders WHERE user_id = 123;

-- Смотрим на "Seq Scan" (плохо) vs "Index Scan" (хорошо)

Проблема: Медленный API запрос

import httpx
import time

async def payment_api_call():
    start = time.time()
    
    async with httpx.AsyncClient() as client:
        # Таймаут 30 секунд (очень долго!)
        response = await client.post(
            "https://api.payment.com/charge",
            json={"amount": 100},
            timeout=30.0
        )
    
    elapsed = time.time() - start
    print(f"API call took: {elapsed:.2f}s")
    return response.json()

# Если API медленный — используем cache
from functools import lru_cache

@lru_cache(maxsize=1000)
async def get_exchange_rate(currency: str) -> float:
    # Кэшируем результат на 1 час
    result = await payment_api_call()
    return result["rate"]

Проблема: Большие данные (Memory)

# БЕЗ оптимизации: загружает всё в памяти
all_data = User.objects.all()  # 1 млн пользователей = 1 ГБ памяти
for user in all_data:
    process(user)

# С оптимизацией: iterator загружает по 1000 за раз
for user in User.objects.all().iterator(chunk_size=1000):
    process(user)  # Используем ~10 МБ вместо 1 ГБ

Шаг 4: Инструменты профилирования

New Relic / DataDog

# Автоматическое отслеживание
# newrelic.ini
[newrelic]
enabled = true
app_name = My App

# Потом видим в dashboard:
# - Slow transactions
# - Database queries
# - External API calls

SQLAlchemy + SQL logging

import logging

# Включаем логирование SQL
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# Теперь видим все SQL запросы с временем
# INFO:sqlalchemy.engine.Engine:SELECT * FROM users
# INFO:sqlalchemy.engine.Engine:() [0.00045s]

Async profiling (asyncio)

import asyncio
import time

# Меряем время async функций
class TimingMiddleware:
    async def __call__(self, scope, receive, send):
        start = time.time()
        await app(scope, receive, send)
        elapsed = time.time() - start
        print(f"Request took {elapsed:.2f}s")

Шаг 5: Типичные решения

Пример: Полная отладка

# БЫЛО: Запрос выполняется 30 секунд
@app.get("/api/users/{user_id}/report")
async def get_user_report(user_id: int):
    # Этап 1: Получить пользователя
    user = await db.query(User).filter(User.id == user_id).first()
    
    # Этап 2: Получить все заказы (N+1 проблема!)
    orders = await db.query(Order).filter(Order.user_id == user_id).all()
    for order in orders:
        order.items  # Отдельный запрос для каждого заказа!
    
    # Этап 3: Считаем статистику (O(n²)!)
    stats = {}
    for order in orders:
        for item in order.items:
            if item.product.category not in stats:
                stats[item.product.category] = 0
            stats[item.product.category] += item.quantity
    
    return {"user": user, "orders": orders, "stats": stats}

# ПОСЛЕ: Оптимизация
@app.get("/api/users/{user_id}/report")
async def get_user_report(user_id: int):
    # 1. Оптимизация: используем join вместо N+1
    user = await db.query(User).options(
        selectinload(User.orders).selectinload(Order.items).selectinload(Item.product)
    ).filter(User.id == user_id).first()
    
    # 2. Оптимизация: SQL для статистики (O(1))
    stats = await db.query(
        Product.category,
        func.sum(Item.quantity).label("total")
    ).filter(Order.user_id == user_id).group_by(Product.category).all()
    
    # Результат: 2 запроса вместо 1000+
    return {"user": user, "orders": user.orders, "stats": stats}

До: 30 сек После: 0.1 сек Улучшение: 300x

Золотое правило

1. ИЗМЕРЬ — что именно медленно?
2. НАЙДИ узкое место (DB, API, Algorithm, Memory)
3. ОПТИМИЗИРУЙ — индексы, кэш, алгоритмы
4. ПРОВЕРЬ — улучшилось ли?
5. МОНИТОРЬ — используй APM (New Relic, DataDog)

Главное: Не гадай. Использование данных и профилирование спасает часы работы.