Как в Python можно получить план выполнения SQL-запроса и фактическое время его выполнения?

Анализ SQL-запросов: EXPLAIN и Timing

Использую несколько методов для анализа производительности SQL-запросов в Python.

1. EXPLAIN для плана выполнения

Bольшинство баз данных поддерживают EXPLAIN для получения плана выполнения.

PostgreSQL

import psycopg2
from psycopg2 import sql

connection = psycopg2.connect("dbname=mydb user=postgres")
cursor = connection.cursor()

# Получение плана выполнения
query = "EXPLAIN ANALYZE SELECT * FROM users WHERE age > 18"
cursor.execute(query)

plan = cursor.fetchall()
for row in plan:
    print(row[0])

cursor.close()
connection.close()

Результат EXPLAIN ANALYZE показывает:

• Тип операции (Seq Scan, Index Scan, Join)
• Стоимость в условных единицах
• Количество возвращённых строк
• Реальное время выполнения

SQLAlchemy с PostgreSQL

from sqlalchemy import create_engine, text

engine = create_engine("postgresql://user:password@localhost/mydb")

with engine.connect() as conn:
    # План выполнения (без анализа)
    result = conn.execute(text("EXPLAIN SELECT * FROM users WHERE age > 18"))
    print("\n".join([row[0] for row in result]))
    
    # План с анализом (выполняет запрос)
    result = conn.execute(text("EXPLAIN ANALYZE SELECT * FROM users WHERE age > 18"))
    print("\n".join([row[0] for row in result]))

2. Измерение времени выполнения

Использование time

import time
import psycopg2

connection = psycopg2.connect("dbname=mydb user=postgres")
cursor = connection.cursor()

start_time = time.time()
cursor.execute("SELECT * FROM users WHERE age > 18")
results = cursor.fetchall()
end_time = time.time()

print(f"Время выполнения: {end_time - start_time:.4f} сек")
print(f"Количество строк: {len(results)}")

cursor.close()
connection.close()

С использованием timeit

import timeit
from sqlalchemy import create_engine, text

engine = create_engine("postgresql://user:password@localhost/mydb")

def execute_query():
    with engine.connect() as conn:
        return conn.execute(text("SELECT * FROM users WHERE age > 18")).fetchall()

# Измеряем среднее время на 100 итерациях
average_time = timeit.timeit(execute_query, number=100) / 100
print(f"Среднее время: {average_time:.4f} сек")

3. Профилирование с Django ORM

from django.db import connection
from django.test.utils import CaptureQueriesContext

# Перехват и анализ запросов
with CaptureQueriesContext(connection) as context:
    # Выполнение запроса
    users = User.objects.filter(age__gt=18).all()
    list(users)  # Материализация

# Анализ
for query in context.captured_queries:
    print(f"SQL: {query[sql]}")
    print(f"Время: {query[time]} сек\n")

4. logging для автоматического отслеживания

import logging
import time
from sqlalchemy import event
from sqlalchemy.engine import Engine

# Настройка логирования медленных запросов
logger = logging.getLogger(sqlalchemy.engine)
logger.setLevel(logging.INFO)

@event.listens_for(Engine, "before_cursor_execute")
def receive_before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault(query_start_time, []).append(time.time())

@event.listens_for(Engine, "after_cursor_execute")
def receive_after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total_time = time.time() - conn.info[query_start_time].pop(-1)
    
    if total_time > 0.1:  # Логируем запросы > 100мс
        logger.warning(f"Slow query ({total_time:.3f}s): {statement}")

5. Контекстный менеджер для профилирования

from contextlib import contextmanager
import time

@contextmanager
def query_timer(description=""):
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{description}: {elapsed:.4f} сек")

# Использование
with query_timer("Fetch users"):
    cursor.execute("SELECT * FROM users WHERE age > 18")
    results = cursor.fetchall()

6. Анализ индексов

# PostgreSQL: проверка использования индексов
def analyze_index_usage(connection, table_name):
    cursor = connection.cursor()
    
    query = f"""
    SELECT 
        schemaname,
        tablename,
        indexname,
        idx_scan as index_scans,
        idx_tup_read as tuples_read,
        idx_tup_fetch as tuples_fetched
    FROM pg_stat_user_indexes
    WHERE tablename = {table_name}
    ORDER BY idx_scan DESC;
    """
    
    cursor.execute(query)
    for row in cursor.fetchall():
        print(row)
    
    cursor.close()

7. QUERY_TIMEOUT для безопасности

from sqlalchemy import create_engine, pool, event

engine = create_engine(
    "postgresql://user:password@localhost/mydb",
    pool_pre_ping=True,  # Проверка соединения перед использованием
)

@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_conn, connection_record):
    # PostgreSQL: установка timeout
    cursor = dbapi_conn.cursor()
    cursor.execute("SET statement_timeout = 5000")
    cursor.close()

Лучшие практики оптимизации

1. Всегда используй EXPLAIN ANALYZE перед оптимизацией
2. Добавляй индексы на часто фильтруемые колонки
3. Мониторь медленные запросы в логах
4. Используй N+1 детектор для SQLAlchemy (автоматическое предупреждение)
5. Кэшируй результаты часто выполняемых запросов
6. Профилируй на реальных данных, не на малых датасетах

Эти методы помогают выявлять и устранять узкие места в работе с БД.