SQLAlchemy в Python: введение в ORM и Core на версии 2.0

SQLAlchemy — это библиотека Python для работы с реляционными базами данных (SQLite, PostgreSQL, MySQL, SQL Server, Oracle и другими). В ней два слоя: Core строит и выполняет SQL-выражения из объектов Python, а ORM поверх Core сопоставляет таблицы с классами и строки с объектами.

Ниже — рабочий пример на SQLAlchemy 2.0 (проверено 23.09.2026 на Python 3.14.6, SQLAlchemy 2.0.54, SQLite 3.53), затем разбор: подключение, модели, запросы, параметры, транзакции и миграции. Код в стиле 1.x (session.query, engine.execute) здесь не используется — почему, объясняю в отдельном разделе.

Пять понятий, которые путают

Связка в одну строку: Engine дает соединения, Session (ORM) или Connection (Core) выполняют запросы в транзакции, модели и MetaData описывают схему.

Установка

python -m pip install SQLAlchemy
python -c "import sqlalchemy; print(sqlalchemy.__version__)"

Вторая команда печатает версию: примеры ниже рассчитаны на ветку 2.x (проверено на 2.0.54).

Для SQLite ничего больше не нужно: драйвер sqlite3 входит в стандартную библиотеку Python. Для других СУБД драйвер ставится отдельно, и он указывается в URL подключения:

Пароль не храните в коде: читайте его из переменной окружения. Если в пароле есть @, / или :, собирайте адрес через sqlalchemy.URL.create(...) — он экранирует такие символы сам.

Минимальный полный пример ORM

Две таблицы — студенты и оценки, связь «один ко многим». Скрипт создает базу, сохраняет студента с двумя оценками и читает их обратно.

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


class Base(DeclarativeBase):
    pass


class Student(Base):
    __tablename__ = "students"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)
    grades: Mapped[list["Grade"]] = relationship(back_populates="student")


class Grade(Base):
    __tablename__ = "grades"

    id: Mapped[int] = mapped_column(primary_key=True)
    course: Mapped[str] = mapped_column(String(100))
    score: Mapped[int]
    student_id: Mapped[int] = mapped_column(ForeignKey("students.id"))
    student: Mapped[Student] = relationship(back_populates="grades")


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

with Session(engine) as session:
    anna = Student(name="Анна", email="anna@example.com")
    anna.grades = [Grade(course="Python", score=5),
                   Grade(course="SQL", score=4)]
    session.add(anna)
    session.commit()

    stmt = select(Student).where(Student.email == "anna@example.com")
    student = session.scalars(stmt).one()
    for g in student.grades:
        print(student.name, g.course, g.score)

При первом запуске в папке появится файл college.db, а в консоли две строки:

Анна Python 5
Анна SQL 4

Разбор по шагам

1. DeclarativeBase — базовый класс, от которого наследуются модели. Через Base.metadata доступен реестр всех таблиц.
2. Mapped[int] — аннотация типа колонки. Mapped[str] без Optional дает NOT NULL, Mapped[str | None] — колонку, допускающую NULL. mapped_column() добавляет детали: ключ, длину, unique.
3. relationship() описывает связь на уровне объектов: anna.grades — список оценок, grade.student — владелец. В базе связь держит внешний ключ student_id.
4. create_all() создает только отсутствующие таблицы. Изменить уже существующую (добавить колонку) он не умеет — для этого нужны миграции.
5. session.add() ставит объект в очередь, commit() отправляет INSERT и фиксирует транзакцию. Оценки сохраняются вместе со студентом: по умолчанию у relationship() включен каскад save-update, и связанные объекты попадают в сессию вместе с родителем.
6. select() строит запрос, session.scalars(...).one() возвращает ровно один объект или бросает исключение, если строк ноль или больше одной.

Чтобы увидеть сгенерированный SQL, передайте create_engine(..., echo=True): каждый запрос и параметры будут печататься в лог.

Запросы: select() вместо session.query()

В 2.0 основной способ строить запрос — функция select(). Одна и та же конструкция работает и в ORM, и в Core. Если напечатать запрос, видно, что значение не вклеено в текст, а передано параметром:

stmt = select(Student.name).where(Student.id == 1)
print(stmt)

SELECT students.name 
FROM students 
WHERE students.id = :id_1

print() показывает обобщенную форму с именованным параметром. При выполнении запрос компилируется под диалект базы: для SQLite параметр станет ?, для PostgreSQL с драйвером psycopg — %(id_1)s::INTEGER (диалект добавляет приведение типа). Точный SQL под свою базу видно через echo=True или print(stmt.compile(engine)). Самые частые операции:

Core: SQL без классов

Core удобен, когда нужны отчеты, массовые вставки или точный контроль над SQL, а объекты не нужны. Таблица описывается через Table и MetaData:

from sqlalchemy import (Column, Integer, MetaData, String, Table,
                        create_engine, insert, select)

engine = create_engine("sqlite://")  # база в памяти
metadata = MetaData()
users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("login", String(50), nullable=False, unique=True),
    Column("role", String(20), nullable=False),
)
metadata.create_all(engine)

with engine.begin() as conn:  # commit при выходе, rollback при исключении
    conn.execute(insert(users), [
        {"login": "anna", "role": "admin"},
        {"login": "boris", "role": "user"},
    ])

with engine.connect() as conn:
    rows = conn.execute(select(users.c.login).where(users.c.role == "user"))
    print(rows.all())

Вывод: [('boris',)]. Разница блоков: engine.begin() сам фиксирует транзакцию, а внутри engine.connect() изменения нужно зафиксировать явным conn.commit(), иначе при закрытии они откатятся.

Сырой SQL и параметры: как не получить инъекцию

SQLAlchemy не защищает от инъекции автоматически, если вы сами собираете строку запроса. Неверный вариант — вставить ввод пользователя в text() через f-строку:

from sqlalchemy import create_engine, text

engine = create_engine("sqlite://")
with engine.begin() as conn:
    conn.execute(text("CREATE TABLE users (login TEXT, role TEXT)"))
    conn.execute(text("INSERT INTO users VALUES ('anna', 'admin'), ('boris', 'user')"))

login = "x' OR '1'='1"  # строка пришла от пользователя

with engine.connect() as conn:
    bad = conn.execute(text(f"SELECT login, role FROM users WHERE login = '{login}'"))
    print("f-строка:", bad.all())

    good = conn.execute(
        text("SELECT login, role FROM users WHERE login = :login"),
        {"login": login},
    )
    print("параметр:", good.all())

f-строка: [('anna', 'admin'), ('boris', 'user')]
параметр: []

С f-строкой условие превратилось в OR '1'='1', и запрос вернул всех пользователей. С именованным параметром :login строка передана драйверу как значение, и пользователя с таким логином нет. Правило: значения — только параметрами. Имена таблиц и колонок параметрами не передаются — их выбирают из белого списка в коде.

Транзакции и откат

Удобный шаблон ORM — session.begin(): при успехе фиксирует транзакцию, при любом исключении внутри блока откатывает ее целиком. Продолжим в том же скрипте, что и первый пример: модели, engine и Session уже определены, Анна сохранена.

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

try:
    with Session(engine) as session, session.begin():
        session.add(Student(name="Борис", email="boris@example.com"))
        session.add(Student(name="Анна 2", email="anna@example.com"))
except IntegrityError as e:
    print("откат:", e.orig)

with Session(engine) as session:
    print("студентов:", session.scalar(select(func.count(Student.id))))

откат: UNIQUE constraint failed: students.email
студентов: 1

Второй email нарушил unique, и Борис тоже не сохранился: обе вставки были в одной транзакции. Текст ошибки здесь от SQLite; у PostgreSQL и MySQL он другой, а класс исключения SQLAlchemy тот же — IntegrityError. Поэтому ловите исключение по типу, а не по тексту.

Миграции: Alembic кратко

Когда схема меняется (новая колонка, индекс), create_all() не поможет. Для этого у авторов SQLAlchemy есть отдельный инструмент Alembic. Базовый цикл:

python -m pip install alembic
alembic init migrations
alembic revision --autogenerate -m "create students"
alembic upgrade head

Перед revision в alembic.ini указывают адрес базы, а в migrations/env.py импортируют Base из модуля с моделями и пишут target_metadata = Base.metadata. Автогенерация сравнивает модели с базой и пишет черновик миграции. Его обязательно читают глазами: переименование колонки, например, может быть распознано как удаление и добавление, то есть с потерей данных.

Что изменилось по сравнению с 1.x

Многие примеры в сети написаны под SQLAlchemy 1.3/1.4 и в 2.0 не работают или устарели:

session.query() в 2.0 оставлен как legacy-интерфейс, но новый код пишут через select().

Если не получилось

• IntegrityError: UNIQUE constraint failed при повторном запуске примера. Скрипт снова вставляет того же студента. Удалите college.db или меняйте email.
• DetachedInstanceError ... is not bound to a Session. Вы обращаетесь к атрибуту объекта после закрытия сессии, а после commit() атрибуты помечены устаревшими. Читайте данные внутри блока with Session(...) или создайте сессию с expire_on_commit=False, понимая, что тогда значения могут устареть.
• Много одинаковых запросов при обходе связей (проблема N+1). По умолчанию student.grades грузится отдельным запросом для каждого студента. На четырех студентах это 5 запросов вместо 2 с select(Student).options(selectinload(Student.grades)).
• ModuleNotFoundError: No module named 'psycopg' при create_engine(). Не установлен драйвер из URL: поставьте его (python -m pip install "psycopg[binary]" для postgresql+psycopg). Если же ошибка NoSuchModuleError: Can't load plugin, в URL опечатка в имени диалекта или драйвера (например, postgresql+psycopg4).
• Новая колонка в модели, а в базе ее нет. create_all() не меняет существующие таблицы — нужна миграция Alembic.

Выводы

• SQLAlchemy — два слоя: Core (SQL-выражения и соединения) и ORM (классы, объекты, Session); ORM построен поверх Core.
• В версии 2.0 модели описывают через DeclarativeBase, Mapped и mapped_column(), запросы — через select().
• Значения в запросах передают только параметрами, в том числе в text(); f-строка с вводом пользователя дает SQL-инъекцию.
• session.begin() и engine.begin() откатывают всю транзакцию при исключении — так связанные изменения сохраняются вместе или не сохраняются вовсе.
• create_all() подходит для учебного старта, изменения схемы в рабочем проекте ведут миграциями Alembic.

Где применяется / связь с практикой

SQLAlchemy используют в бэкендах на FastAPI и Flask, в сервисах обработки данных и в скриптах, которым нужна переносимость между SQLite на ноутбуке и PostgreSQL на сервере. Следующие шаги после этой статьи — связь «многие ко многим», стратегии загрузки связей, асинхронный режим sqlalchemy.ext.asyncio и тестирование кода, работающего с базой.

FAQ

Нужно ли знать SQL, чтобы работать с SQLAlchemy ORM?
Да, хотя бы основы: SELECT, JOIN, индексы и транзакции. ORM генерирует SQL за вас, но без понимания запросов трудно заметить лишние обращения к базе и медленные соединения.

Работает ли SQLAlchemy с asyncio?
Да, в 2.0 есть create_async_engine и AsyncSession в модуле sqlalchemy.ext.asyncio. Нужен асинхронный драйвер, например asyncpg или psycopg в асинхронном режиме для PostgreSQL, aiosqlite для SQLite.

Чем SQLAlchemy отличается от Django ORM?
Django ORM встроен во фреймворк Django и работает внутри него. SQLAlchemy — самостоятельная библиотека: ее подключают к любому приложению, и у нее есть отдельный слой Core для SQL без моделей.