📖 Теория: SQLAlchemy — CRUD и запросы

1. CRUD — что это такое

В рамках работы с базами данных взаимодействие с данными осуществляется через четыре основные операции:

• C — Create (Создание) — добавление новых записей
• R — Read (Чтение) — получение данных
• U — Update (Обновление) — изменение существующих записей
• D — Delete (Удаление) — удаление записей

Эти операции формируют базовый интерфейс между приложениями и хранилищами данных, обеспечивая структурированный подход к управлению данными. Для большинства моделей базы данных реализуются все четыре операции.

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

engine = create_engine('sqlite:///example.db')

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int] = mapped_column(Integer)

Base.metadata.create_all(engine)

2. Create — создание записей

Создание новых записей начинается с создания экземпляра модели. Затем экземпляр добавляется в сессию и фиксируется в базе данных.

Добавление одного объекта

with Session(engine) as session:
    new_user = User(name="Alice", age=30)
    session.add(new_user)
    session.commit()

Добавление нескольких объектов сразу

with Session(engine) as session:
    session.add_all([
        User(name='Bob',   age=22),
        User(name='David', age=27),
        User(name='Alice', age=30),
        User(name='Ann',   age=17),
        User(name='Ann',   age=27),
    ])
    session.commit()

3. Read — чтение данных

Чтение данных выполняется через различные запросы — от простого доступа по первичному ключу до сложной выборки с фильтрацией.

Чтение по первичному ключу

with Session(engine) as session:
    # session.get() — получить по первичному ключу (современный 2.x)
    user = session.get(User, 1)
    if user:
        print(user.name, user.age)

Выборка всех записей (SQLAlchemy 2.x)

from sqlalchemy import select

with Session(engine) as session:
    users = session.scalars(select(User)).all()
    for user in users:
        print(user.name, user.age)

Выборка с фильтрацией

with Session(engine) as session:
    stmt = select(User).where(User.name == "Alice")
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name)

4. Update — обновление данных

Обновление включает получение объекта из базы, изменение его атрибутов и вызов commit(). SQLAlchemy автоматически отслеживает изменения объектов в сессии.

with Session(engine) as session:
    user = session.get(User, 1)
    if user:
        user.age = 35       # изменяем возраст
        session.commit()    # SQLAlchemy автоматически генерирует UPDATE

    # Проверяем результат
    user = session.get(User, 1)
    print(user.name, user.age)  # → Alice 35

5. Delete — удаление данных

Удаление объекта выполняется через session.delete(obj) с последующим commit().

with Session(engine) as session:
    user = session.get(User, 1)
    if user:
        session.delete(user)
        session.commit()
        print("Deleted")
    else:
        print("User with id 1 isn't found")

6. Построение запросов — методы извлечения

SQLAlchemy предоставляет гибкие инструменты для построения запросов, включая фильтрацию, сортировку, группировку, присоединения и подзапросы.

Запросы строятся через конструктор select(), который затем передаётся в session.execute() или session.scalars(). Результат уточняется методами-терминаторами.

Методы извлечения результата

.all() — список всех объектов

Возвращает список всех объектов, соответствующих запросу.

with Session(engine) as session:
    users = session.scalars(select(User)).all()
    for user in users:
        print(user.name, user.age)

.first() — первый объект или None

Возвращает первый объект из результата запроса или None, если результат пуст. Никогда не выбрасывает исключение.

with Session(engine) as session:
    user = session.scalars(select(User)).first()
    if user:
        print(user.id, user.name)

.one() — ровно один объект

Возвращает ровно один объект. Если результатов нет или их больше одного — генерирует исключение (NoResultFound или MultipleResultsFound).

with Session(engine) as session:
    stmt = select(User).where(User.id == 3)
    user = session.scalars(stmt).one()
    print(user)

.one_or_none() — один или None

Возвращает один объект или None, если объектов нет. Если объектов более одного — генерирует исключение.

with Session(engine) as session:
    stmt = select(User).where(User.id == 3)
    user = session.scalars(stmt).one_or_none()
    print(user)

7. Фильтрация результатов

Фильтрация выполняется через метод .where() (2.x) или .filter() (1.x). Это аналог оператора WHERE в SQL.

Базовая фильтрация

with Session(engine) as session:
    # Выборка пользователей старше 25 лет
    stmt = select(User).where(User.age > 25)
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)

Функции фильтрации

Кроме классических операторов сравнения используются специальные функции, применяемые к полям модели:

like() — поиск по шаблону

with Session(engine) as session:
    # Имена, начинающиеся на "A"
    stmt = select(User).where(User.name.like('A%'))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name)

between() — диапазон значений

with Session(engine) as session:
    # Пользователи с ID от 2 до 4
    stmt = select(User).where(User.id.between(2, 4))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name)

in_() — принадлежность списку

with Session(engine) as session:
    names = ["Alice", "Bob"]
    stmt = select(User).where(User.name.in_(names))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name)

8. Логические условия

SQLAlchemy позволяет использовать логические операторы для комбинирования условий фильтрации:

from sqlalchemy import and_, or_, not_

with Session(engine) as session:
    # AND: пользователи старше 20 и моложе 23
    stmt = select(User).where(and_(User.age > 20, User.age < 23))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)

    # OR: старше 30 или имя David
    stmt = select(User).where(or_(User.age > 30, User.name == 'David'))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)

    # NOT: не имя David
    stmt = select(User).where(not_(User.name == 'David'))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)

# Эквивалентно and_(User.age > 20, User.age < 23)
stmt = select(User).where(User.age > 20, User.age < 23)

9. Сортировка результатов

Сортировка выполняется через метод .order_by(). По умолчанию — по возрастанию. Для убывания используется функция desc().

from sqlalchemy import desc

with Session(engine) as session:
    # Сортировка по возрасту (по возрастанию)
    stmt = select(User).order_by(User.age)
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)

    # Сортировка по возрасту (по убыванию)
    stmt = select(User).order_by(desc(User.age))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)

    # Многоуровневая: возраст убывает, имя по алфавиту
    stmt = select(User).order_by(desc(User.age), User.name)
    users = session.scalars(stmt).all()
    for user in users:
        print(user.id, user.name, user.age)