Выполнение CRUD (Create, Read, Update, Delete) операций через SQLAlchemy

Давайте вспоминать, что такое CRUD—Создание (Create), чтение (Read), обновление (Update) и удаление (Delete) — это основные операции, которые мы выполняем при работе с базой данных. Эти четыре операции составляют основу любого веб-приложения, от блогов до интернет-магазинов.

Вот пример из реальной жизни. Представьте, что вы менеджер в книжном магазине. Вы добавляете новые книги в каталог (Create), проверяете, есть ли нужная книга в наличии (Read), обновляете цену книги (Update) и убираете из каталога книги, которые уже много лет как распроданы (Delete).

SQLAlchemy как раз помогает нам стать такими себе "менеджерами" для базы данных, но без тяжёлых коробок.

---

Основы работы с CRUD в SQLAlchemy

У SQLAlchemy есть удобный интерфейс для выполнения CRUD-операций. Центральным элементом в нём является класс Session, который управляет всеми взаимодействиями с базой данных.

Session работает как посредник между нашими моделями и реальной базой данных. Сессия заботится о выполнении транзакций и поддержании консистентности данных.

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Создаём движок
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)

# Создаём фабрику сессий
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Сессия готова. Теперь мы можем использовать её для операций с базой данных.

---

Создание записей (Create)

Чтобы добавить новую запись в базу, нужно:

1. Создать объект модели с необходимыми данными.
2. Добавить его в текущую сессию.
3. Зафиксировать изменения (закоммитить).

Вот пример добавления нового пользователя в таблицу users:

from models import User  # Предположим, вы уже создали модель User

# Создаём сессию
db = SessionLocal()

# Создаём объект пользователя
new_user = User(name="Иван Иванов", email="ivan.ivanov@example.com")

# Добавляем его в сессию
db.add(new_user)

# Фиксируем изменения
db.commit()

# Закрываем сессию
db.close()

После выполнения этого кода новая запись появится в таблице users. Обратите внимание на db.commit() — без него изменения не сохранятся.

---

Чтение записей (Read)

Чтение данных в SQLAlchemy — это то, где начинается магия. Вы можете выбирать отдельные записи, фильтровать их по параметрам или даже собирать агрегированные данные.

1. Запрос всех записей

users = db.query(User).all()
for user in users:
    print(user.name, user.email)

db.query(User).all() вернёт все записи в таблице users в виде списка.

2. Фильтрация данных

Теперь добавим немного изящества с фильтрацией. Например, выберем пользователя с определённым email:

user = db.query(User).filter(User.email == "ivan.ivanov@example.com").first()
if user:
    print(f"Найден пользователь: {user.name}")
else:
    print("Пользователь не найден")

Здесь мы использовали метод .filter() и передали в него условие. Если таких пользователей несколько, используем .all() вместо .first().

3. Использование операторов

SQLAlchemy поддерживает множество операторов для фильтрации. Вот несколько примеров:

• Равенство: User.name == "Иван"
• Больше/меньше: User.age > 30
• LIKE: User.name.like("И%")
• IN: User.id.in_([1, 2, 3])

Все эти операторы можно объединять с помощью and_, or_ из sqlalchemy:

from sqlalchemy import and_, or_

users = db.query(User).filter(
    or_(
        User.name == "Иван",
        User.email.like("%example.com")
    )
).all()

---

Обновление записей (Update)

Для обновления записей используется комбинация query() и метода update() или обновление через экземпляр модели.

Обновление через экземпляр модели

user = db.query(User).filter(User.email == "ivan.ivanov@example.com").first()
if user:
    user.name = "Иван Петров"
    db.commit()  # Не забудьте зафиксировать изменения

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

Обновление через метод update()

Этот способ обновляет данные напрямую:

db.query(User).filter(User.email == "ivan.ivanov@example.com").update({User.name: "Иван Петров"})
db.commit()

Если планируете обновлять сразу много записей, update() работает быстрее.

---

Удаление записей (Delete)

Самый драматичный из CRUD-операций — удаление. Чтобы удалить запись:

1. Находим нужный объект через query().
2. Вызываем метод delete().

Удаление одной записи

user = db.query(User).filter(User.email == "ivan.ivanov@example.com").first()
if user:
    db.delete(user)
    db.commit()

Удаление сразу нескольких записей

db.query(User).filter(User.name.like("И%")).delete()
db.commit()

Имейте в виду, что массовое удаление может быть опасным. Всегда проверяйте свои фильтры. Уловка для параноиков: сначала сделайте select() и убедитесь, что выбираете только нужные записи.

---

Обработка ошибок при CRUD операциях

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

1. Используйте try-except

from sqlalchemy.exc import SQLAlchemyError

try:
    user = db.query(User).filter(User.id == 1).first()
    if user:
        db.delete(user)
        db.commit()
except SQLAlchemyError as e:
    print(f"Произошла ошибка: {e}")
    db.rollback()  # Откатываем изменения

2. Откат транзакций

Если что-то пошло не так, вы можете откатить транзакцию с помощью db.rollback().

---

Полезные ссылки

• Официальная документация по SQLAlchemy
• Руководство по ORM в SQLAlchemy
• Шпаргалка по написанию запросов