SQLAlchemy Core: SQL из Python-выражений

Собрать запрос через f-строку — первое, что приходит в голову: f"SELECT * FROM tasks WHERE title = '{name}'". Работает ровно до того дня, когда в name прилетает апостроф или чужой кусок SQL. За удобством прячутся две проблемы сырых строк.

Первая: SQL живёт в строке, и любая опечатка или неаккуратная вставка пользовательских данных это потенциальная SQL-инъекция. Параметры ? спасают, но про них нужно помнить каждый раз.

Вторая: каждая СУБД имеет свой диалект SQL. Если приложение пишется под SQLite, а потом переезжает на PostgreSQL — почти наверняка часть запросов придётся переписывать.

SQLAlchemy Core решает обе проблемы: SQL строится из Python-выражений, безопасность встроена по умолчанию, и один и тот же код работает с PostgreSQL, MySQL, SQLite. Простые SELECT/WHERE действительно похожи во всех СУБД, но как только заходим в специфические функции (даты, строки, агрегаты) или схему, синтаксис расходится, и Core переводит ваш Python в правильный диалект:

Иллюстрация: вверху Python-выражение select(tasks).where(extract("year", tasks.c.created_at) == 2026), стрелка к SQLAlchemy Core, ниже три варианта SQL — у PostgreSQL и MySQL одинаковое EXTRACT(year FROM created_at), а у SQLite другое CAST(STRFTIME('%Y', created_at) AS INTEGER); плейсхолдер значения у каждой СУБД свой: %(param_1)s, %s, ?

Установка


pip install sqlalchemy

Для SQLite дополнительных драйверов не нужно. Для PostgreSQL ставится отдельно psycopg2-binary, для MySQL — pymysql.

Engine: подключение

Engine это объект, отвечающий за связь с БД. Создаётся один раз на приложение:

Python 3.13
1234
from sqlalchemy import create_engine

engine = create_engine('sqlite:///core_tasks.db', echo=True)
print("Engine готов")
Engine готов

Параметр echo=True включает вывод выполняемых SQL-запросов в консоль. Удобно в обучении и при отладке, в production его выключают.

Строка подключения для других СУБД:

postgresql://user:pass@host:5432/dbname
mysql+pymysql://user:pass@host/dbname
sqlite:///file.db

Описание таблицы

В Core структура таблицы описывается объектом Table — Python-эквивалент SQL-команды CREATE TABLE:

Python 3.13
1234567891011121314151617
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Boolean

engine = create_engine('sqlite:///core_tasks.db')
metadata = MetaData()

tasks_table = Table(
    'tasks',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('title', String, nullable=False),
    Column('completed', Boolean, default=False),
)

# Создаём таблицу в БД (если её ещё нет)
metadata.create_all(engine)

print("Таблица tasks готова")
Таблица tasks готова

MetaData это коллекция всех Table-объектов приложения. metadata.create_all(engine) создаёт сразу все таблицы из коллекции, которых ещё нет в БД.

Те же четыре операции, но выражениями

Те же CREATE, READ, UPDATE, DELETE, что и в прошлой главе, — только вместо SQL-строк Python-выражения. Для каждой операции в Core есть помощник: insert(), select(), update(), delete().

Знакомую четвёрку пробегаем быстро, отмечая новое: пакетную вставку списком, доступ к колонкам через .c и Python-операторы в .where(). Дальше предполагаем, что engine и tasks_table уже определены, как выше.

INSERT

Python 3.13
1234567891011121314
from sqlalchemy import insert

with engine.connect() as connection:
    result = connection.execute(
        insert(tasks_table),
        [
            {'title': 'Изучить SQLAlchemy Core', 'completed': False},
            {'title': 'Написать приложение', 'completed': False},
            {'title': 'Протестировать код', 'completed': False},
        ],
    )
    connection.commit()

print(f"Добавлено строк: {result.rowcount}")
Добавлено строк: 3

Значения передаются списком словарей — это batch-вставка одним запросом. SQLAlchemy сам подставит параметры безопасно.

SELECT

Python 3.13
123456
from sqlalchemy import select

with engine.connect() as connection:
    result = connection.execute(select(tasks_table))
    for row in result:
        print(row.id, row.title, row.completed)
1 Изучить SQLAlchemy Core False
2 Написать приложение False
3 Протестировать код False

Доступ к колонкам по имени (row.title), а не по индексу как в sqlite3. И completed вернулся как False, а не 0: SQLAlchemy знает тип колонки (Boolean) и сам приводит значение к Python-типу — в сыром sqlite3 в этом месте было бы 0. Условие фильтрации добавляется через .where():

Python 3.13
12345678
from sqlalchemy import select

with engine.connect() as connection:
    result = connection.execute(
        select(tasks_table).where(tasks_table.c.id == 1)
    )
    row = result.first()
    print(row.title)
Изучить SQLAlchemy Core

tasks_table.c.id это «колонка id таблицы tasks». Сравнения (==, >, <, .in_(), .like()) превращаются в SQL автоматически.

UPDATE

Python 3.13
1234567891011
from sqlalchemy import update

with engine.connect() as connection:
    result = connection.execute(
        update(tasks_table)
        .where(tasks_table.c.id == 1)
        .values(completed=True)
    )
    connection.commit()

print(f"Обновлено строк: {result.rowcount}")
Обновлено строк: 1

DELETE

Python 3.13
123456789
from sqlalchemy import delete

with engine.connect() as connection:
    result = connection.execute(
        delete(tasks_table).where(tasks_table.c.id == 3)
    )
    connection.commit()

print(f"Удалено строк: {result.rowcount}")
Удалено строк: 1

Проверка понимания

Главное преимущество SQLAlchemy Core перед сырыми SQL-строками в sqlite3?

В следующей статье возьмём SQLAlchemy ORM — слой выше Core, где таблицы становятся Python-классами, строки — объектами, и вам почти не нужно думать в терминах SQL. Хорошо подходит для типичной бизнес-логики; Core остаётся в арсенале для случаев, когда нужен точный контроль над запросом.

Telegram