💻 Примеры: CRUD и запросы (SQLAlchemy 2.x)

Пример 0: Подготовка модели и тестовых данных

Во всех примерах используется модель User и набор данных из лекции.

from sqlalchemy import create_engine, select, String, Integer, and_, or_, not_, desc
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# Движок — SQLite в памяти (удобно для обучения)
engine = create_engine("sqlite:///:memory:", echo=False)

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)

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

Base.metadata.create_all(engine)

# Тестовые данные из лекции
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()
print("Данные добавлены.")

Пример 1: Create — добавление пользователей

# Добавление одного пользователя
with Session(engine) as session:
    new_user = User(name="Alice", age=30)
    session.add(new_user)
    session.commit()
    print(f"Создан: {new_user.id}, {new_user.name}")  # id присваивается после commit

# Добавление нескольких пользователей
with Session(engine) as session:
    session.add_all([
        User(name='Bob',   age=22),
        User(name='David', age=27),
    ])
    session.commit()

Пример 2: Read — чтение по первичному ключу

with Session(engine) as session:
    user = session.get(User, 1)   # по id=1
    if user:
        print(user.name, user.age)
    else:
        print("Пользователь не найден")

Пример 3: Read — выборка всех записей

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

Пример 4: Update — обновление данных

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 — удаление пользователя

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: Методы извлечения — .all(), .first(), .one(), .one_or_none()

with Session(engine) as session:
    # .all() — список всех
    users = session.scalars(select(User)).all()
    print(".all():", users)

    # .first() — первый или None
    user = session.scalars(select(User)).first()
    print(".first():", user)

    # .one() — ровно один (выбросит исключение при 0 или >1)
    stmt = select(User).where(User.id == 3)
    user = session.scalars(stmt).one()
    print(".one():", user)

    # .one_or_none() — один или None (выбросит при >1)
    stmt = select(User).where(User.id == 3)
    user = session.scalars(stmt).one_or_none()
    print(".one_or_none():", user)

Пример 7: Фильтрация — сравнения

with Session(engine) as session:
    # Пользователи старше 25 лет
    stmt = select(User).where(User.age > 25)
    users = session.scalars(stmt).all()
    print("Старше 25:")
    for user in users:
        print(f"  {user.id}, {user.name}, {user.age}")

Пример 8: Фильтрация — like(), between(), in_()

with Session(engine) as session:
    # like: имена на "A"
    stmt = select(User).where(User.name.like('A%'))
    users = session.scalars(stmt).all()
    print("Имена на A:", [u.name for u in users])

    # between: id от 2 до 4
    stmt = select(User).where(User.id.between(2, 4))
    users = session.scalars(stmt).all()
    print("ID 2-4:", [(u.id, u.name) for u in users])

    # in_: имена из списка
    names = ["Alice", "Bob"]
    stmt = select(User).where(User.name.in_(names))
    users = session.scalars(stmt).all()
    print("Alice или Bob:", [u.name for u in users])

Пример 9: Логические условия — 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()
    print("Возраст 20-23:", users)

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

    # NOT: не David
    stmt = select(User).where(not_(User.name == 'David'))
    users = session.scalars(stmt).all()
    print("Не David:", users)

Пример 10: Сортировка

with Session(engine) as session:
    # Возраст: по возрастанию
    stmt = select(User).order_by(User.age)
    users = session.scalars(stmt).all()
    print("По возрасту (ASC):")
    for u in users:
        print(f"  {u.id}, {u.name}, {u.age}")

    # Возраст: по убыванию
    stmt = select(User).order_by(desc(User.age))
    users = session.scalars(stmt).all()
    print("По возрасту (DESC):")
    for u in users:
        print(f"  {u.id}, {u.name}, {u.age}")

    # Многоуровневая: возраст DESC, имя ASC
    stmt = select(User).order_by(desc(User.age), User.name)
    users = session.scalars(stmt).all()
    print("Возраст DESC + имя ASC:")
    for u in users:
        print(f"  {u.id}, {u.name}, {u.age}")