В каком свойстве можно посмотреть текст SQL запроса

Просмотр текста SQL запроса в Python

Это очень практичный вопрос, который часто встаёт при отладке баз данных. Рассмотрим различные подходы в зависимости от ORM и драйвера.

1. SQLAlchemy (самый популярный ORM в Python)

Способ 1: Превращение query в строку

from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import Session

engine = create_engine('postgresql://user:pass@localhost/db')

# ❌ НЕПРАВИЛЬНО: просто вывести query
query = select(User).where(User.id == 1)
print(query)  # <SELECT sqlalchemy.sql.selectable.Select object at 0x...>

# ✅ ПРАВИЛЬНО: использовать str() или compile()
query = select(User).where(User.id == 1)
print(str(query))  # Выведет SQL (но с привязками как параметры)

# Правильнее: скомпилировать с диалектом БД
compiled = query.compile()
print(compiled)  # SELECT user.id, user.name FROM user WHERE user.id = ?

Способ 2: compile() с параметрами

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from sqlalchemy.sql import compiler

# Получить SQL с параметрами
query = select(User).where(User.id == 1)
compiled = query.compile()

# Вариант 1: просто строка запроса
print(compiled)  # SELECT user.id, user.name FROM user WHERE user.id = ?

# Вариант 2: с параметрами
print(compiled.params)  # {'id_1': 1}

# Вариант 3: скомпилировать с конкретным диалектом
from sqlalchemy.dialects import postgresql
compiled = query.compile(dialect=postgresql.dialect())
print(compiled)  # SELECT "user".id, "user".name FROM "user" WHERE "user".id = %(id_1)s

Способ 3: compile с параметрами для конкретной БД

from sqlalchemy import create_engine, select
from sqlalchemy.dialects import postgresql, mysql

query = select(User).where(User.id == 1)

# PostgreSQL
postgres_query = str(query.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
print(postgres_query)  # SELECT user.id, user.name FROM user WHERE user.id = 1

# MySQL
mysql_query = str(query.compile(dialect=mysql.dialect()))
print(mysql_query)  # SELECT user.id, user.name FROM user WHERE user.id = %s

# SQLite
from sqlalchemy.dialects import sqlite
sqlite_query = str(query.compile(dialect=sqlite.dialect()))
print(sqlite_query)  # SELECT user.id, user.name FROM user WHERE user.id = ?

Способ 4: Логирование в Echo mode (лучший для отладки)

from sqlalchemy import create_engine

# ✅ ПРАВИЛЬНО: включить echo=True при создании engine
engine = create_engine(
    'postgresql://user:pass@localhost/db',
    echo=True  # Выводит все SQL запросы в лог
)

# Теперь все запросы будут выведены
with Session(engine) as session:
    users = session.query(User).filter(User.id == 1).all()
    # В логе увидишь:
    # SELECT user.id, user.name FROM "user" WHERE user.id = %s
    # [SQL parameters: (1,)]

Способ 5: Логирование SQLAlchemy на уровне логов

import logging
from sqlalchemy import create_engine

# Включить логирование SQLAlchemy
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

engine = create_engine('postgresql://user:pass@localhost/db')

# Теперь в консоли увидишь все запросы
with Session(engine) as session:
    users = session.query(User).filter(User.id == 1).all()
    # INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
    # INFO:sqlalchemy.engine.Engine:SELECT user.id, user.name FROM "user"
    # INFO:sqlalchemy.engine.Engine:[SQL parameters: (1,)]

Полный пример для отладки

from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import Session
from sqlalchemy.dialects import postgresql

class QueryDebugger:
    """Помощник для отладки SQL запросов"""
    
    @staticmethod
    def show_sql(query, dialect='postgresql'):
        """Вывести SQL запрос с параметрами"""
        dialects = {
            'postgresql': postgresql.dialect(),
            'mysql': mysql.dialect(),
            'sqlite': sqlite.dialect(),
        }
        
        compiled = query.compile(
            dialect=dialects.get(dialect),
            compile_kwargs={"literal_binds": False}  # Не подставлять значения
        )
        
        print(f"SQL Query:")
        print(f"  {compiled}")
        print(f"Parameters:")
        print(f"  {compiled.params}")
        
        return str(compiled)

# Использование
query = select(User).where(User.id == 1).order_by(User.name)
QueryDebugger.show_sql(query)
# Output:
# SQL Query:
#   SELECT user.id, user.name FROM user WHERE user.id = :id_1 ORDER BY user.name
# Parameters:
#   {'id_1': 1}

2. Django ORM (если используешь Django)

from django.db import connection
from django.conf import settings

# Способ 1: Просмотр последнего запроса
from django.db import connection

users = User.objects.filter(id=1)
print(str(users.query))  # SELECT "user"."id", "user"."name" FROM "user" WHERE "user"."id" = 1

# Способ 2: Просмотр всех запросов (только в DEBUG mode)
if settings.DEBUG:
    for query in connection.queries:
        print(query['sql'])
        print(query['time'])  # Время выполнения

# Способ 3: использовать django-debug-toolbar
# Добавляет визуальный интерфейс для просмотра SQL запросов

3. Raw SQL with sqlite3 / psycopg2 / pymysql

import sqlite3
from typing import Any

# В этих драйверах ты уже сам пишешь SQL, поэтому всё просто
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()

# Вот твой SQL запрос - видно сразу
sql_query = "SELECT id, name FROM users WHERE id = ?"
print(sql_query)  # SELECT id, name FROM users WHERE id = ?

# С параметрами
params = (1,)
cursor.execute(sql_query, params)
# Полный запрос: SELECT id, name FROM users WHERE id = 1

4. SQLModel (FastAPI + SQLAlchemy)

from sqlalchemy import select
from sqlmodel import SQLModel, Field, Session

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

# Те же методы как в SQLAlchemy
query = select(User).where(User.id == 1)
print(str(query.compile()))  # SELECT user.id, user.name, user.email FROM user WHERE user.id = ?

5. Практические примеры отладки

Пример 1: Отладка производительности

import time
from sqlalchemy import create_engine, event, select
from sqlalchemy.orm import Session

engine = create_engine('postgresql://user:pass@localhost/db')

# Логирование времени выполнения
@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())
    print(f"Query started: {statement}")
    print(f"Parameters: {parameters}")

@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)
    print(f"Query completed in {total_time:.3f} seconds\n")

# Теперь каждый запрос будет залогирован
with Session(engine) as session:
    users = session.query(User).all()

Пример 2: Форматирование SQL для читаемости

from sqlalchemy import select
import sqlparse

query = select(User).join(Profile).where(User.id == 1)
compiled = str(query.compile())

# Отформатировать SQL для читаемости
formatted_sql = sqlparse.format(
    compiled,
    reindent=True,
    keyword_case='upper'
)
print(formatted_sql)
# Output:
# SELECT user.id, user.name, profile.bio
# FROM user
# JOIN profile ON user.id = profile.user_id
# WHERE user.id = ?

Пример 3: Проверка запроса перед выполнением

from sqlalchemy import select, text
from sqlalchemy.orm import Session

class QueryValidator:
    @staticmethod
    def validate_before_execute(session: Session, query):
        """Проверить query перед выполнением"""
        compiled = query.compile()
        sql_str = str(compiled)
        
        print(f"Query to execute:")
        print(sql_str)
        print(f"\nParameters: {compiled.params}")
        
        # Проверки
        if 'DROP' in sql_str.upper():
            raise ValueError("Dangerous DROP query!")
        if 'DELETE' in sql_str.upper() and 'WHERE' not in sql_str:
            raise ValueError("DELETE without WHERE clause!")
        
        # Выполнить если всё ОК
        return session.execute(query).all()

# Использование
with Session(engine) as session:
    query = select(User).where(User.id == 1)
    results = QueryValidator.validate_before_execute(session, query)

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

7. Best practices для отладки

# ✅ ХОРОШО: Логирование с контекстом
def find_user(user_id: int):
    query = select(User).where(User.id == user_id)
    logger.debug(f"Executing query: {query.compile()}")
    
    try:
        result = session.execute(query).scalar_one_or_none()
        logger.info(f"Found user: {result}")
        return result
    except Exception as e:
        logger.error(f"Query failed: {query.compile()}", exc_info=True)
        raise

# ✅ ХОРОШО: В development использовать echo=True
if DEBUG:
    engine = create_engine(DATABASE_URL, echo=True)
else:
    engine = create_engine(DATABASE_URL, echo=False)

# ✅ ХОРОШО: Добавлять explain для медленных запросов
query = select(User).where(User.id == 1)
result = session.execute(text(f"EXPLAIN {str(query.compile())}")).fetchall()
print(result)  # План выполнения запроса

Итоговый ответ

Основной способ посмотреть SQL запрос:

# SQLAlchemy
query = select(User).where(User.id == 1)
print(str(query.compile()))  # Вот SQL запрос

# Или для production отладки
engine = create_engine(db_url, echo=True)  # Выводит все запросы

Ключевые свойства:

• query.compile() — получить скомпилированный объект
• str() — преобразовать в строку
• .params — параметры запроса
• echo=True в engine — логирование всех запросов
• logging.getLogger('sqlalchemy.engine') — для production логирования

Это критично для отладки N+1 queries, оптимизации производительности и общего понимания того, что происходит в БД!