Сырой SQL: text() внутри ORM

Тринадцать уроков мы не написали ни одного SQL-запроса — и это была честная демонстрация силы ORM. Но границы у неё есть: тебе достался готовый SQL от DBA, понадобилась оконная функция, у которой нет выражения в SQLAlchemy, или экзотическая конструкция конкретной СУБД. Хороший инструмент не запирает дверь, а оставляет служебный вход: в SQLAlchemy это text() — настоящий SQL-текст внутри обычной сессии.

Идея текстовых запросов простая: оборачиваешь SQL-строку в text() и исполняешь как обычно — той же сессией, в той же транзакции. Значения подставляются параметрами, результат приходит строками, а не объектами моделей. Сегодня разберём, как это делается аккуратно: с параметрами, без инъекций и с пониманием, где raw уместен, а где ORM по-прежнему лучше.

Почему это вообще работает так гладко? Потому что у ORM и raw одно горло: session.execute принимает любой «исполняемый» объект — и собранный из select(), и text(). Ниже конвейера ORM-выражений лежит тот же драйвер, что получает твой SQL-текст. Поэтому всё, чему ты научился про сессии в уроках 5–8, относится и к сырым запросам: транзакции, commit, rollback — ничего нового учить не придётся.

text(): SQL как есть

Минимальный пример: выбрать выполненные задачи сырым SQL. Обрати внимание на :d в тексте запроса — это именованный параметр, значение для него передаётся словарём вторым аргументом execute. Ровно тот же подход к условиям, что и в ORM-запросах из урока 7: значения никогда не встраиваются в текст.

from sqlalchemy import create_engine, String, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="раз", done=1), Task(title="два"), Task(title="три", done=1)])
    s.commit()

    rows = s.execute(
        text("SELECT id, title FROM tasks WHERE done = :d ORDER BY id"),
        {"d": 1},
    )
    for row in rows:
        print(row.id, "-", row.title)

1 - раз
3 - три

Сессия отработала как обычно: та же транзакция, то же соединение, тот же commit уже случился ранее. Изменился только формат результата — не объекты Task, а строки Row. К ним применимы оба доступа: по имени колонки, row.title, и по индексу, row[0]. Имя берётся из SQL: переименуешь колонку в запросе через AS — переименуется и атрибут строки. Так работают все text()-выборки, и к этому формату стоит привыкнуть сразу: он ближе к DBAPI, чем к ORM, — компромисс за право писать SQL руками.

Параметр :d — единственный правильный способ подставить значение. Драйвер базы сам заключит его в кавычки и экранирует: значение со злой шуткой внутри останется просто строкой. Склейка SQL-текста со значениями — статья про SQL-инъекции, и text() в этом смысле так же безопасен, как ORM-запросы, — если параметры используются по назначению.

from sqlalchemy import create_engine, String, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="отчёт"), Task(title="доработать отчёт"), Task(title="письмо")])
    s.commit()

    rows = s.execute(
        text("SELECT title FROM tasks WHERE title LIKE :p ORDER BY title"),
        {"p": "%от%"},
    ).all()
    print([r.title for r in rows])

['доработать отчёт', 'отчёт']

Заметь, где проходит граница параметра: :p подставляет значение, а значит и шаблон %от% — целиком. Нельзя параметризовать кусок шаблона внутри LIKE, но никто не мешает собрать шаблон в Python и передать параметром целиком — это по-прежнему данные, а не текст запроса.

Агрегаты из сырого SQL читаются через знакомый scalar: запрос «сколько» возвращает одну ячейку, и session.scalar достанет её без разборов строк. Это ровно тот же приём подсчёта из урока 12, только SQL написан руками: count, сумма, максимум — любой одиночный вопрос базе в одну поездку.

from sqlalchemy import create_engine, String, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([Task(title="раз", done=1), Task(title="два"), Task(title="три", done=1)])
    s.commit()

    total = s.scalar(
        text("SELECT COUNT(*) FROM tasks WHERE done = :d"),
        {"d": 0},
    )
    print("невыполненных:", total)

невыполненных: 1

Писать тоже можно: INSERT и UPDATE

text() исполняет любой SQL, не только выборки. Массовая вставка списком словарей — executemany — и точечное обновление выглядят так.

from sqlalchemy import create_engine, String, text, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    done: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add(Task(title="раз", done=1))
    s.commit()

    s.execute(
        text("INSERT INTO tasks (title, done) VALUES (:t, :d)"),
        [{"t": "два", "d": 0}, {"t": "три", "d": 0}],
    )
    s.execute(
        text("UPDATE tasks SET done = :d WHERE title = :t"),
        {"d": 1, "t": "три"},
    )
    s.commit()

    print([t.title for t in s.scalars(
        select(Task).where(Task.done == 1).order_by(Task.id)
    )])

['раз', 'три']

Смотри, что произошло: raw-INSERT доложил две строки, raw-UPDATE пометил одну выполненной, а прочитали мы результат обычной ORM-выборкой — и она честно увидела чужие строки. Raw и ORM живут в одной базе и одной транзакции, поэтому свободно чередуются: пишешь служебную операцию сырым SQL — ORM замечает её после commit, когда перечитывает данные. Список словарей во втором аргументе превратил один execute в executemany: драйвер выполнил INSERT по разу на каждый словарь.

Когда raw оправдан

Теперь — о границах. Список ситуаций, где text() не «костыль», а правильный инструмент.

• запрос уже написан: унаследован от старой системы, прислан DBA, отлажен и работает — переписывать его в ORM ради единообразия дорого и рискованно;
• фича СУБД без аналога в ORM: оконные функции старых версий, специфичные типы и расширения конкретной базы;
• точная форма запроса важна: план запроса профилирован с этим SQL, и любое перефразирование ORM меняет производительность;
• административные операции: ANALYZE, VACUUM, копирование между таблицами — служебный SQL, для которого ORM-выражений просто нет.

from sqlalchemy import create_engine, String, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    points: Mapped[int] = mapped_column(default=0)

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    s.add_all([
        Task(title="отчет", points=6),
        Task(title="верстка", points=8),
        Task(title="багрепорт", points=3),
        Task(title="письмо", points=2),
    ])
    s.commit()

    rows = s.execute(
        text(
            "WITH heavy AS (SELECT title, points FROM tasks WHERE points >= :m) "
            "SELECT title, points FROM heavy ORDER BY points DESC"
        ),
        {"m": 5},
    ).all()
    for row in rows:
        print(row.title, "-", row.points)

верстка - 8
отчет - 6

CTE — общее табличное выражение — вычисляет «виртуальную таблицу» heavy и выбирает уже из неё. SQL-запрос вставлен целиком, как он выглядел бы в консоли базы, с одним параметром порога. Именно такие многоходовки — причина №1, по которой разработчики открывают text(): ORM умеет CTE и сам, но написанный кем-то тяжёлый отчёт проще исполнить, чем переводить на другой язык запросов.

И зеркальный список: где raw не нужен. Опечатка в имени колонки в text() вскроется только в рантайме — ORM-выражения проверяются при построении запроса. Сырой SQL не компонуется: condition добавить в готовую строку — значит склеивать текст, а это дорога к инъекциям. Переносимость тоже страдает: диалектные тонкости SQL-текста поедут с ним на другую СУБД. Если запрос выражается средствами ORM — выражай его ORM; text() для остальных случаев.

from sqlalchemy import create_engine, ForeignKey, String, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user")

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    s.add_all([anna, boris])
    s.flush()
    s.add_all([Task(title="отчёт", user=anna), Task(title="презентация", user=boris)])
    s.commit()

    rows = s.execute(
        text(
            "SELECT u.name, t.title FROM tasks AS t "
            "JOIN users AS u ON t.user_id = u.id ORDER BY t.id"
        )
    ).all()
    for r in rows:
        print(r.name, "-", r.title)

Анна - отчёт
Борис - презентация

Тот же отчёт, что собирался join-ом в уроке 11, только написанный руками с алиасами t и u — в стиле, которым его прислали бы из чужого проекта. ORM-модели при этом живы: данные писались обычным ORM-кодом, читались сырым SQL, и никто не заметил подмены. Для базы эти миры неразличимы — различие только в том, кто формулирует запрос.

Мост между мирами

Частый компромисс: SQL уже написан, но работать с результатом хочется как с объектами. SQLAlchemy склеивает и это: select(Model).from_statement(text(...)) исполняет сырой SQL, а строки превращает в модели, сопоставляя колонки по именам.

from sqlalchemy import create_engine, ForeignKey, String, select, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40))
    tasks = relationship("Task", back_populates="user")

    def __repr__(self):
        return f"User({self.name})"

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(80))
    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"))
    user = relationship("User", back_populates="tasks")

engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)

with Session(engine) as s:
    anna = User(name="Анна")
    boris = User(name="Борис")
    s.add_all([anna, boris])
    s.flush()
    s.add_all([
        Task(title="отчёт", user=anna),
        Task(title="презентация", user=boris),
    ])
    s.commit()

    users = s.scalars(
        select(User).from_statement(
            text("SELECT * FROM users WHERE name = :n")
        ),
        {"n": "Борис"},
    ).all()
    print(users)

[User(Борис)]

На выходе — настоящий объект User: с рабочими атрибутами, relationship и отслеживанием сессии. Сырой SQL исполнился, а ORM сопоставил колонки по именам и собрал модель. Это удобный мост для отчётных запросов: текст не трогаем, но и не выпадаем из объектного мира. Держи в голове условие: колонки в SQL должны покрывать поля модели — маппинг идёт по именам. Пропущенная колонка не уронит запрос, но соответствующий атрибут объекта окажется незагруженным и обратится к базе при первом чтении — ленивая загрузка из урока 9 работает и здесь.

# ОПАСНО: значение вклеивается в текст запроса
query = text(f"SELECT * FROM users WHERE name = '{name}'")

# безопасно: значение едет параметром, драйвер экранирует сам
query = text("SELECT * FROM users WHERE name = :n")
rows = session.execute(query, {"n": name})

# то же правило для кусков SQL: строки склеивает SQL, а не Python
query = text("SELECT title FROM tasks ORDER BY :col")   # так нельзя
query = text("SELECT title FROM tasks ORDER BY " + col) # и так нельзя

Что дальше

Теперь ORM незамкнута: рядом с типизированными выражениями легально живёт настоящий SQL — в параметрах, через from_statement, с моделями на выходе. Девять уроков строили арсенал, десятый показал клапан для исключений — это зрелая форма любого инструмента: и правило, и выход из него.

Дальше курс выходит за пределы песочницы: SQLAlchemy во Flask, где сессией управляет фреймворк, тестирование ORM через pytest на in-memory базе, миграции Alembic и переезд с SQLite на PostgreSQL. Сырой SQL из сегодняшнего урока встретится там не раз: в тестах им проверяют точные состояния таблицы, в миграциях — переписывают данные при смене схемы.

text() — служебный вход, а не запасная дверь: тот же движок, та же сессия, тот же драйвер с параметрами. Знаешь, зачем тебе настоящий SQL, — входи смело.