SQLAlchemy: ORM и работа с базами данных в Python

SQLAlchemy — главная библиотека Python для работы с реляционными базами данных: SQLite, PostgreSQL, MySQL и другими. Она позволяет описать таблицы как классы Python (ORM) и писать запросы методами вместо строк SQL. Минимальный пример SQLAlchemy на Python в стиле 2.0:

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


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))


engine = create_engine("sqlite:///app.db")  # файл app.db появится рядом со скриптом
Base.metadata.create_all(engine)            # создать таблицы

with Session(engine) as session:
    session.add(User(name="Анна"))
    session.commit()
    print(session.scalars(select(User.name)).all())  # → ['Анна']

Ниже — установка, подключение к базе, модели, сессия, запросы, связи между таблицами и частые ошибки. Примеры проверены на SQLAlchemy 2.1 и Python 3.12 с базой SQLite; синтаксис совпадает с SQLAlchemy 2.0.

Установка SQLAlchemy

Пакет ставится через pip, лучше в виртуальное окружение проекта:

# Windows
py -m pip install sqlalchemy

# macOS / Linux
python3 -m pip install sqlalchemy

Проверка версии:

import sqlalchemy

print(sqlalchemy.__version__)  # → 2.1.4

SQLAlchemy 2.1 требует Python 3.11 или новее. Для SQLite больше ничего не нужно: драйвер sqlite3 входит в стандартную библиотеку Python. Для других СУБД ставится драйвер отдельно — например, psycopg для PostgreSQL или pymysql для MySQL. Подробнее об установке пакетов — в руководстве по pip.

Важно: в интернете много примеров в старом стиле 1.x — с declarative_base(), Column(...) и session.query(User). В SQLAlchemy 2.x они в основном ещё работают, но новый код пишите в стиле 2.0: DeclarativeBase, Mapped, mapped_column и select(). Именно его использует эта статья.

Engine: подключение к базе данных

Engine — точка входа: он знает, к какой базе подключаться, и управляет пулом соединений. Создаётся один раз на всё приложение функцией create_engine() с URL базы данных:

from sqlalchemy import create_engine

engine = create_engine("sqlite:///app.db")    # файл app.db в текущей папке
engine = create_engine("sqlite://")           # SQLite в памяти — для тестов и экспериментов
engine = create_engine(
    "postgresql+psycopg://user:password@localhost:5432/mydb"
)                                             # PostgreSQL через драйвер psycopg 3

Формат URL: диалект+драйвер://пользователь:пароль@хост:порт/база. Для SQLite путь к файлу идёт после трёх слешей; абсолютный путь в Linux и macOS даёт четыре слеша подряд: sqlite:////home/user/app.db.

Параметр echo=True печатает в консоль весь SQL, который отправляет SQLAlchemy, — удобно при обучении и отладке:

engine = create_engine("sqlite://", echo=True)

Модели: DeclarativeBase, Mapped и mapped_column

Модель — это класс Python, который описывает таблицу. Все модели наследуются от общего базового класса, унаследованного от DeclarativeBase:

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


class Base(DeclarativeBase):
    pass


class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)
    bio: Mapped[str | None]  # необязательное поле: NULL разрешён

    books: Mapped[list["Book"]] = relationship(
        back_populates="author", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"Author(id={self.id}, name={self.name!r})"


class Book(Base):
    __tablename__ = "books"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    year: Mapped[int]
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))

    author: Mapped["Author"] = relationship(back_populates="books")

    def __repr__(self) -> str:
        return f"Book(title={self.title!r}, year={self.year})"

Как это читать:

• __tablename__ — имя таблицы в базе.
• Аннотация Mapped[...] задаёт тип Python, а по нему — тип столбца: int → INTEGER, str → VARCHAR, datetime → DATETIME и т. д.
• Mapped[str] означает NOT NULL, а Mapped[str | None] — столбец, допускающий NULL.
• mapped_column() нужен, когда требуются дополнительные параметры: primary_key, unique, index, default, длина строки через String(100), внешний ключ ForeignKey. Если параметров нет, хватит одной аннотации, как у поля year.

Целочисленный первичный ключ id заполняется базой автоматически. Если вы раньше работали с моделями Django, идея та же, отличается синтаксис.

Python: создание базы данных и таблиц

Чтобы создать базу данных и все таблицы по моделям, вызовите create_all():

from sqlalchemy import create_engine

engine = create_engine("sqlite:///library.db", echo=True)
Base.metadata.create_all(engine)

Для SQLite файл library.db создаётся автоматически при первом подключении. С echo=True видно, какой SQL выполнила библиотека:

CREATE TABLE authors (
	id INTEGER NOT NULL,
	name VARCHAR(100) NOT NULL,
	bio VARCHAR,
	PRIMARY KEY (id),
	UNIQUE (name)
)

CREATE TABLE books (
	id INTEGER NOT NULL,
	title VARCHAR(200) NOT NULL,
	year INTEGER NOT NULL,
	author_id INTEGER NOT NULL,
	PRIMARY KEY (id),
	FOREIGN KEY(author_id) REFERENCES authors (id)
)

create_all() создаёт только отсутствующие таблицы и не меняет уже существующие. Если вы добавили поле в модель, существующая таблица не обновится — для изменения схемы в реальных проектах используют миграции через Alembic (инструмент от авторов SQLAlchemy). Сервер PostgreSQL или MySQL create_all() тоже не создаёт: базу на сервере заводят заранее, а библиотека создаёт в ней таблицы.

Session: добавление, изменение и удаление

Все операции с объектами идут через Session — она отслеживает изменения и отправляет их в базу при commit(). Сессию открывают в блоке with, чтобы она гарантированно закрылась:

from sqlalchemy.orm import Session

with Session(engine) as session:
    pushkin = Author(name="Пушкин")
    pushkin.books = [
        Book(title="Евгений Онегин", year=1833),
        Book(title="Капитанская дочка", year=1836),
    ]
    tolstoy = Author(name="Толстой", books=[Book(title="Война и мир", year=1869)])

    session.add_all([pushkin, tolstoy])
    session.commit()
    print(pushkin.id, tolstoy.id)  # → 1 2

Книги сохранились вместе с авторами: связанные объекты добавляются в сессию автоматически. После commit() у объектов появились id, выданные базой.

Изменение и удаление — обычная работа с объектами плюс commit():

with Session(engine) as session:
    book = session.scalars(select(Book).where(Book.title == "Война и мир")).one()
    book.year = 1867         # изменение отслеживается сессией
    session.commit()         # → UPDATE books SET year=? WHERE books.id = ?

    session.delete(book)
    session.commit()         # → DELETE FROM books WHERE books.id = ?

Если что-то пошло не так, session.rollback() отменяет незакоммиченные изменения. Если сессия закрывается без commit(), изменения тоже не сохраняются.

Для веб-приложений обычно создают фабрику сессий через sessionmaker. Блок with SessionLocal.begin() сам вызовет commit() в конце или rollback() при исключении:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(engine)

with SessionLocal.begin() as session:
    session.add(Author(name="Гоголь"))
# здесь изменения уже закоммичены

Запросы: select(), where и order_by

В стиле 2.0 запрос строится функцией select(), а выполняется через сессию. Для получения объектов моделей используйте session.scalars():

from sqlalchemy import select

with Session(engine) as session:
    stmt = select(Book).where(Book.year < 1850).order_by(Book.year)
    for book in session.scalars(stmt):
        print(book, book.author.name)
# → Book(title='Евгений Онегин', year=1833) Пушкин
# → Book(title='Капитанская дочка', year=1836) Пушкин

Как получить результат:

Вызов	Что возвращает
session.scalars(stmt).all()	список объектов
session.scalars(stmt).first()	первый объект или None
session.scalars(stmt).one()	ровно один объект, иначе исключение
session.scalars(stmt).one_or_none()	один объект или None; если строк больше одной — исключение
session.scalar(stmt)	первое значение первой строки (удобно для count)
session.get(Author, 1)	объект по первичному ключу или None
session.execute(stmt)	строки-кортежи, когда выбираете несколько столбцов

Частые условия фильтрации:

select(Book).where(Book.year >= 1830, Book.year < 1840)   # несколько условий — это AND
select(Book).where(Book.title.like("%дочка%"))            # LIKE
select(Book).where(Book.year.in_([1833, 1869]))           # IN
select(Author).where(Author.bio.is_(None))                # IS NULL
select(Book).order_by(Book.year.desc()).limit(10)         # сортировка и LIMIT

Для OR есть функция or_() из пакета sqlalchemy. Агрегаты — через func:

from sqlalchemy import func

stmt = (
    select(Author.name, func.count(Book.id))
    .join(Author.books)
    .group_by(Author.name)
    .order_by(Author.name)
)
with Session(engine) as session:
    for name, count in session.execute(stmt):
        print(name, count)
# → Пушкин 2
# → Толстой 1

Массовые изменения без загрузки объектов делаются конструкциями update() и delete():

from sqlalchemy import delete, update

with Session(engine) as session:
    session.execute(update(Book).where(Book.year < 1850).values(year=Book.year + 1))
    session.execute(delete(Book).where(Book.year > 1900))
    session.commit()

Связи между таблицами: relationship

В моделях выше связь «один ко многим» задана двумя частями: внешним ключом author_id в таблице книг и атрибутами relationship() с обеих сторон. Параметр back_populates связывает их между собой, поэтому book.author и author.books всегда согласованы:

with Session(engine) as session:
    author = session.scalars(select(Author).where(Author.name == "Толстой")).one()
    print(author.books)       # → [Book(title='Война и мир', year=1869)]
    print(author.books[0].author is author)  # → True

Параметр cascade="all, delete-orphan" у Author.books означает: при удалении автора удаляются и его книги, а книга, убранная из списка author.books, удаляется из базы.

По умолчанию связанные объекты загружаются «лениво» — отдельным запросом при первом обращении. Если вы перебираете сотню авторов и у каждого читаете books, получится сто дополнительных запросов (проблема N+1). Решение — загрузить связи заранее через selectinload:

from sqlalchemy.orm import selectinload

with Session(engine) as session:
    stmt = select(Author).options(selectinload(Author.books))
    for author in session.scalars(stmt):
        print(author.name, [b.title for b in author.books])  # без лишних запросов

Подробно о проблеме N+1 на примере Django — в статье про оптимизацию запросов ORM.

Сырой SQL через text()

Иногда проще написать SQL вручную. Для этого есть text(); значения передавайте через именованные параметры, а не форматированием строк:

from sqlalchemy import text

with Session(engine) as session:
    rows = session.execute(
        text("SELECT title FROM books WHERE year < :year"), {"year": 1850}
    )
    print(rows.scalars().all())

Параметры защищают от SQL-инъекций — никогда не подставляйте пользовательский ввод в SQL через f-строки.

Альтернатива без ORM: модуль sqlite3

Если задача — просто создать базу данных на Python и сохранить несколько таблиц, SQLAlchemy не обязательна. Встроенный модуль sqlite3 работает без установки пакетов:

import sqlite3

conn = sqlite3.connect("library.db")  # файл создастся, если его нет
with conn:  # транзакция: commit при успехе, rollback при ошибке
    conn.execute(
        "CREATE TABLE IF NOT EXISTS books ("
        "id INTEGER PRIMARY KEY, title TEXT NOT NULL, year INTEGER)"
    )
    conn.executemany(
        "INSERT INTO books (title, year) VALUES (?, ?)",
        [("Евгений Онегин", 1833), ("Война и мир", 1869)],
    )

for row in conn.execute("SELECT title, year FROM books WHERE year < ?", (1850,)):
    print(row)  # → ('Евгений Онегин', 1833)
conn.close()

Обратите внимание: with conn управляет транзакцией, но не закрывает соединение — close() вызывается отдельно.

Задача	Что выбрать
Небольшой скрипт, одна-две таблицы, только SQLite	sqlite3
Модели как классы, связи, переход между СУБД	SQLAlchemy ORM
API на FastAPI или Flask с базой данных	SQLAlchemy (или SQLModel)
Сайт на Django	встроенный Django ORM

SQLAlchemy и веб-фреймворки

В Django своя ORM, и SQLAlchemy там обычно не используют. А вот в FastAPI и Flask SQLAlchemy — стандартный выбор для работы с базой. Как подключить её к FastAPI, выдавать сессию через зависимость и организовать CRUD, показано в статье «FastAPI и SQLAlchemy».

У SQLAlchemy есть и асинхронный режим (create_async_engine, AsyncSession из sqlalchemy.ext.asyncio) для асинхронных драйверов вроде asyncpg или aiosqlite. Модели и select() в нём те же, меняется способ выполнения запросов (await session.execute(...)). Начиная с SQLAlchemy 2.1 нужная для этого библиотека greenlet не ставится автоматически — устанавливайте пакет командой pip install "sqlalchemy[asyncio]".

Частые ошибки

DetachedInstanceError после закрытия сессии

DetachedInstanceError: Parent instance <Author at 0x...> is not bound to a Session;
lazy load operation of attribute 'books' cannot proceed

Объект загрузили в одной сессии, сессию закрыли, а потом обратились к ленивой связи. Читайте связанные данные внутри блока with Session(...) или загрузите их заранее через selectinload.

IntegrityError: UNIQUE constraint failed

Вы вставляете значение, которое нарушает ограничение unique, NOT NULL или внешний ключ. После ошибки сессия находится в состоянии отката — вызовите session.rollback(), прежде чем продолжать работу с ней:

from sqlalchemy.exc import IntegrityError

try:
    session.add(Author(name="Пушкин"))
    session.commit()
except IntegrityError:
    session.rollback()

Изменения не сохранились в базе

Забыт session.commit(). Сессия, закрытая без коммита, откатывает изменения.

Новое поле модели не появилось в таблице

create_all() не изменяет существующие таблицы. Для учебной базы SQLite можно удалить файл и создать её заново; в рабочем проекте используйте миграции Alembic.

ModuleNotFoundError: No module named ‘sqlalchemy’

Пакет установлен в другой интерпретатор, чем тот, которым запускается код. Решение — в статье про ModuleNotFoundError.

Частые вопросы

Как сделать связь «многие ко многим»?

Нужна промежуточная таблица с двумя внешними ключами. Её объявляют через Table без отдельной модели и передают в relationship(secondary=...):

from sqlalchemy import Column, ForeignKey, Table

book_tags = Table(
    "book_tags",
    Base.metadata,
    Column("book_id", ForeignKey("books.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True),
)


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)
    books: Mapped[list["Book"]] = relationship(secondary=book_tags, back_populates="tags")

# в модели Book:
#     tags: Mapped[list["Tag"]] = relationship(secondary=book_tags, back_populates="books")

Строки в book_tags SQLAlchemy добавляет и удаляет сама, когда вы меняете списки book.tags или tag.books.

Как хранить в столбце значение из Enum?

Укажите класс перечисления Python в аннотации — SQLAlchemy создаст столбец нужного типа и будет возвращать члены перечисления:

import enum


class Status(enum.Enum):
    draft = "draft"
    published = "published"


class Book(Base):
    # ...
    status: Mapped[Status] = mapped_column(default=Status.draft)

В SQLite такой столбец хранится как VARCHAR с именами членов, в PostgreSQL создаётся собственный тип ENUM. Новые значения в перечислении для PostgreSQL потребуют миграции.

Как быстро вставить тысячи строк?

Не создавайте объекты по одному в цикле с commit() на каждой итерации. Передайте список словарей в insert() — SQLAlchemy отправит их пакетом:

from sqlalchemy import insert

with Session(engine) as session:
    session.execute(
        insert(Book),
        [{"title": f"Книга {i}", "year": 2000 + i, "author_id": 1} for i in range(1000)],
    )
    session.commit()

Как увидеть SQL, который сгенерирует запрос?

Напечатайте сам запрос — без выполнения и подключения к базе:

stmt = select(Book.title).where(Book.year < 1850)
print(stmt)
# SELECT books.title
# FROM books
# WHERE books.year < :year_1

print(stmt.compile(compile_kwargs={"literal_binds": True}))  # с подставленными значениями

Чтобы видеть все запросы, которые реально уходят в базу, создайте движок с echo=True.

Как подключить миграции Alembic?

Установите пакет и создайте папку миграций:

pip install alembic
alembic init migrations

В alembic.ini укажите URL базы (sqlalchemy.url), а в migrations/env.py — метаданные моделей: target_metadata = Base.metadata. После этого alembic revision --autogenerate -m "add bio" сравнит модели с базой и создаст миграцию, а alembic upgrade head применит её. Пустая миграция обычно означает, что target_metadata не задан или модули с моделями не импортированы в env.py.

Итог

Для работы с SQLAlchemy в стиле 2.0 достаточно пяти понятий: create_engine() для подключения, модели на DeclarativeBase с Mapped и mapped_column, create_all() для создания таблиц, Session с commit() и запросы через select(). Для простых скриптов база SQLite создаётся и встроенным модулем sqlite3. Следующий шаг — связка FastAPI и SQLAlchemy или PostgreSQL в Django; другие популярные пакеты собраны в обзоре библиотек Python, полный справочник — в документации SQLAlchemy.