Raw SQL в Django ORM | Курс Django ORM урок 7.1

Цель урока

Разобрать инструменты для работы с сырым SQL в Django: Manager.raw(), connection.cursor(). Понять когда ORM не справляется и нужен прямой SQL. Освоить параметризованные запросы как единственный безопасный способ передачи данных.

Необходимые знания

Урок 2.2: QuerySet API
Урок 3.3: annotate, aggregate
Базовое понимание SQL injection

Когда ORM недостаточно

ORM покрывает 95% задач. Но есть ситуации когда сырой SQL проще или единственный вариант:

Сложные рекурсивные запросы (CTE, WITH RECURSIVE)
Специфичные функции PostgreSQL без обертки
Оптимизированные запросы которые ORM генерирует неэффективно
Массовые операции со специфическим SQL (COPY, INSERT ... ON CONFLICT)
Вызов хранимых процедур

Manager.raw() - SELECT с маппингом на модель

raw() выполняет SQL SELECT и возвращает RawQuerySet, объекты модели, как обычный QuerySet, но без возможности дальнейшей фильтрации через ORM.

from shop.models import Product

# Простой raw запрос
products = Product.objects.raw("SELECT * FROM shop_product WHERE is_active = true")

for p in products:
    print(p.name, p.price)  # атрибуты модели доступны

SELECT * FROM shop_product WHERE is_active = true
-- выполняется как есть

Объекты p, настоящие экземпляры Product. Можно вызывать методы модели, обращаться к полям.

Параметризованные запросы - обязательно

Никогда не форматируйте SQL строкой Python. Это SQL injection:

# ОПАСНО - SQL injection!
name = request.GET.get("name")
products = Product.objects.raw(f"SELECT * FROM shop_product WHERE name = '{name}'")
# Если name = "'; DROP TABLE shop_product; --" то это катастрофа

Всегда используйте параметризованные запросы:

# Безопасно, параметры передаются отдельно
name = request.GET.get("name")
products = Product.objects.raw(
    "SELECT * FROM shop_product WHERE name = %s",
    [name]
)

-- Django передает в PostgreSQL как:
SELECT * FROM shop_product WHERE name = $1
-- с параметром name отдельно
-- PostgreSQL никогда не интерпретирует значение параметра как SQL

Параметры передаются списком [value] или словарем {"key": value} в зависимости от синтаксиса:

# Список, позиционные параметры
Product.objects.raw(
    "SELECT * FROM shop_product WHERE price > %s AND stock >= %s",
    [min_price, min_stock]
)

# Словарь, именованные параметры (PostgreSQL синтаксис)
Product.objects.raw(
    "SELECT * FROM shop_product WHERE price > %(min_price)s AND stock >= %(min_stock)s",
    {"min_price": min_price, "min_stock": min_stock}
)

Обязательные поля в raw()

raw() требует чтобы запрос возвращал первичный ключ модели. Django использует его для создания объектов:

# Ошибка, нет pk, но исключение возникает при итерации, не при вызове raw()
qs = Product.objects.raw("SELECT name, price FROM shop_product")
list(qs)  # FieldDoesNotExist: Raw query must include the primary key

# Правильно, включить id
Product.objects.raw("SELECT id, name, price FROM shop_product")

Дополнительные поля через raw()

raw() может возвращать поля которых нет в модели, они доступны как атрибуты объекта:

products = Product.objects.raw("""
    SELECT shop_product.id, shop_product.name, shop_product.price,
           COUNT(shop_review.id) AS review_count
    FROM shop_product
    LEFT JOIN shop_review ON shop_product.id = shop_review.product_id
    GROUP BY shop_product.id, shop_product.name, shop_product.price
    ORDER BY review_count DESC
    LIMIT 10
""")

for p in products:
    print(f"{p.name}: {p.review_count} отзывов")  # review_count доступен

Это работает для простых добавочных полей. Для сложных агрегаций удобнее annotate() в ORM.

translations - маппинг имен столбцов

Если SQL возвращает столбцы с именами отличными от полей модели, translations говорит Django какой столбец соответствует какому полю:

products = Product.objects.raw(
    "SELECT id, name AS product_name, price AS product_price FROM shop_product LIMIT 3",
    translations={"product_name": "name", "product_price": "price"}
)

for p in products:
    print(p.name, p.price)  # атрибуты модели доступны через исходные имена полей

translations словарь: ключ это имя столбца в SQL (алиас), а значение, имя поля модели.

Когда нужен translations, это унаследованная таблица с другими именами столбцов или представление (VIEW) с отличающимися именами.

connection.cursor() - произвольный SQL

cursor() дает прямой доступ к DBAPI курсору PostgreSQL. Подходит для любого SQL: SELECT, INSERT, UPDATE, DELETE, DDL.

from django.db import connection

def get_category_stats():
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                c.name,
                COUNT(p.id) AS product_count,
                AVG(p.price) AS avg_price,
                SUM(p.stock) AS total_stock
            FROM shop_category c
            LEFT JOIN shop_product p ON c.id = p.category_id
            GROUP BY c.id, c.name
            ORDER BY product_count DESC
        """)
        # fetchall() возвращает список кортежей
        rows = cursor.fetchall()

    # rows = [("Phones", 15, 799.50, 150), ("Laptops", 8, 1299.00, 40), ...]
    return [
        {
            "name": row[0],
            "product_count": row[1],
            "avg_price": row[2],
            "total_stock": row[3],
        }
        for row in rows
    ]

SELECT c.name, COUNT(p.id) AS product_count, ...
FROM shop_category c
LEFT JOIN shop_product p ON c.id = p.category_id
GROUP BY c.id, c.name
ORDER BY product_count DESC;

fetchone, fetchmany, fetchall

with connection.cursor() as cursor:
    cursor.execute("SELECT COUNT(*) FROM shop_product WHERE is_active = true")

    # fetchone() - одна строка (кортеж)
    row = cursor.fetchone()
    count = row[0]

    # fetchmany(n) - следующие n строк
    cursor.execute("SELECT id, name FROM shop_product ORDER BY name")
    first_ten = cursor.fetchmany(10)

    # fetchall() - все оставшиеся строки
    cursor.execute("SELECT id, name FROM shop_product ORDER BY name LIMIT 100")
    all_rows = cursor.fetchall()

cursor.description - метаданные столбцов

with connection.cursor() as cursor:
    cursor.execute("SELECT id, name, price FROM shop_product LIMIT 1")
    cursor.fetchall()

    # Имена столбцов
    columns = [col[0] for col in cursor.description]
    # ["id", "name", "price"]

Это полезно для динамического маппинга результатов:

def dict_fetchall(cursor):
    """Вернуть все строки как список словарей."""
    columns = [col[0] for col in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]

with connection.cursor() as cursor:
    cursor.execute("SELECT id, name, price FROM shop_product")
    results = dict_fetchall(cursor)
# [{"id": 1, "name": "iPhone 15", "price": Decimal("999.00")}, ...]

Параметризованные запросы в cursor

Правило то же, что и в raw(), параметры передаются отдельно, никогда не форматировать строку:

# ОПАСНО
cursor.execute(f"UPDATE shop_product SET price = {new_price} WHERE id = {product_id}")

# Безопасно
cursor.execute(
    "UPDATE shop_product SET price = %s, updated_at = NOW() WHERE id = %s",
    [new_price, product_id]
)

UPDATE shop_product SET price = $1, updated_at = NOW() WHERE id = $2
-- $1 = new_price, $2 = product_id

cursor.mogrify() - посмотреть итоговый SQL

mogrify() возвращает строку SQL с подставленными параметрами, без выполнения запроса. Полезно для отладки:

from django.db import connection

with connection.cursor() as cursor:
    sql = cursor.mogrify(
        "SELECT * FROM shop_product WHERE price > %s AND category_id = %s",
        [500, 3]
    )
    print(sql)
    # b"SELECT * FROM shop_product WHERE price > 500 AND category_id = 3"

mogrify() возвращает bytes. Это метод курсора psycopg (PostgreSQL), поэтому работает только с бэкендом PostgreSQL. В psycopg 3 mogrify() доступен через ClientCursor, стандартный курсор его не предоставляет.

Применение, убедиться что параметры подставляются правильно перед выполнением тяжелого запроса:

with connection.cursor() as cursor:
    params = [min_price, max_price, category_id, limit]
    sql_template = """
        SELECT id, name, price
        FROM shop_product
        WHERE price BETWEEN %s AND %s AND category_id = %s
        ORDER BY price
        LIMIT %s
    """
    print(cursor.mogrify(sql_template, params))  # проверяем
    cursor.execute(sql_template, params)          # выполняем

Пример: рекурсивный CTE

ORM не поддерживает WITH RECURSIVE. Для рекурсивных запросов, только raw SQL:

def get_category_tree(root_id):
    """Получить все подкатегории включая вложенные."""
    with connection.cursor() as cursor:
        cursor.execute("""
            WITH RECURSIVE category_tree AS (
                -- базовый случай: корневая категория
                SELECT id, name, parent_id, 0 AS depth
                FROM shop_category
                WHERE id = %s

                UNION ALL

                -- рекурсивный случай: дочерние категории
                SELECT c.id, c.name, c.parent_id, ct.depth + 1
                FROM shop_category c
                INNER JOIN category_tree ct ON c.parent_id = ct.id
            )
            SELECT id, name, depth FROM category_tree ORDER BY depth, name
        """, [root_id])

        return dict_fetchall(cursor)

WITH RECURSIVE category_tree AS (
    SELECT id, name, parent_id, 0 AS depth
    FROM shop_category WHERE id = 1

    UNION ALL

    SELECT c.id, c.name, c.parent_id, ct.depth + 1
    FROM shop_category c
    INNER JOIN category_tree ct ON c.parent_id = ct.id
)
SELECT id, name, depth FROM category_tree ORDER BY depth, name;

Транзакции и cursor()

cursor() выполняется в текущей транзакции. С AUTOCOMMIT=True, автоматически:

with connection.cursor() as cursor:
    cursor.execute("UPDATE shop_product SET stock = stock - 1 WHERE id = %s", [product_id])
# COMMIT автоматически при выходе из cursor()

Для явных транзакций:

from django.db import transaction

with transaction.atomic():
    with connection.cursor() as cursor:
        cursor.execute(
            "UPDATE shop_product SET stock = stock - %s WHERE id = %s",
            [quantity, product_id]
        )
        # COMMIT произойдет при выходе из atomic()

callproc() - хранимые процедуры

with connection.cursor() as cursor:
    cursor.callproc("recalculate_product_stats", [product_id])
    result = cursor.fetchone()

CALL recalculate_product_stats(1);

Когда raw() vs cursor()

	`raw()`	`cursor()`
Тип запроса	Только SELECT	Любой SQL
Результат	Объекты модели	Кортежи / словари
Дополнительная логика	Нет	Полный контроль
Когда использовать	SELECT с маппингом на модель	DDL, процедуры, сложные запросы

Практическое задание

1) Напишите функцию get_top_products_by_revenue(limit=10) через cursor(). Запрос: продукты с наибольшей суммой выручки из OrderItem (sum of price * quantity). Верните список словарей.

2) Реализуйте get_category_with_products_raw(category_slug) через Manager.raw(). Запрос должен возвращать объекты Product с дополнительным полем review_count (количество отзывов).

3) Напишите функцию get_category_tree(root_id), рекурсивный CTE. Возвращайте список категорий с полем depth (глубина вложенности).

4) Напишите безопасную функцию поиска search_products(query) через cursor(). Запрос должен искать по name и description через LIKE. Убедитесь что параметры передаются безопасно (параметризованный запрос).

5) Замерьте время выполнения одного и того же запроса (топ-10 продуктов по количеству отзывов) тремя способами:

Через ORM (annotate(Count("reviews")).order_by("-review_count")[:10])
Через raw()
Через cursor()

Сравните читаемость и производительность.

Возможные ошибки

Конкатенация параметров в SQL

# SQL INJECTION - никогда так не делать
status = request.POST.get("status")
cursor.execute(f"UPDATE shop_order SET status = '{status}'")

# Правильно
cursor.execute("UPDATE shop_order SET status = %s WHERE id = %s", [status, order_id])

PostgreSQL библиотека psycopg экранирует параметры. Форматирование строкой обходит защиту.

raw() без первичного ключа

# FieldDoesNotExist при итерации (не при вызове raw())
qs = Product.objects.raw("SELECT name, price FROM shop_product")
list(qs)  # FieldDoesNotExist: Raw query must include the primary key

# Правильно
Product.objects.raw("SELECT id, name, price FROM shop_product")

Использование cursor() для запросов которые решаются ORM

# Излишне сложно
with connection.cursor() as cursor:
    cursor.execute("SELECT COUNT(*) FROM shop_product WHERE is_active = true")
    count = cursor.fetchone()[0]

# Проще
count = Product.objects.filter(is_active=True).count()

Raw SQL нужен только когда ORM не справляется. Для стандартных CRUD операций, всегда ORM.

Забыть закрыть cursor

# Утечка ресурсов
cursor = connection.cursor()
cursor.execute("SELECT ...")
# cursor не закрыт!

# Правильно, контекстный менеджер закрывает автоматически
with connection.cursor() as cursor:
    cursor.execute("SELECT ...")

Связь со следующим уроком

В уроке 7.2 разберем специфичные для PostgreSQL возможности Django ORM: JSONField, ArrayField и полнотекстовый поиск. Это инструменты которые позволяют использовать PostgreSQL как документную БД и полнотекстовый поисковик прямо в Django ORM.

<< Урок 6.3

Урок 7.2 >>

Подписывайтесь на мой Telegram канал

Если вам нужен ментор и вы хотите научиться веб-разработке, узнать подробнее