📖 Теория: конспект блока SQLAlchemy (Уроки 05–07)

SQLAlchemy: ORM vs Core

SQLAlchemy — библиотека Python для работы с реляционными базами данных. Предоставляет два уровня:

Преимущества ORM: абстракция, легкость, интеграция. Недостатки: производительность при сложных запросах, неявное поведение.

Преимущества Core: скорость, полный контроль над SQL. Недостатки: больше кода, ниже читаемость.

Движок (Engine) и строка подключения

Строка подключения имеет формат:

'<DBMS>+<library>://<user>:<password>@<host>:<port>/<database>'

Примеры:

from sqlalchemy import create_engine

# PostgreSQL
engine = create_engine('postgresql://user:password@localhost:5432/mydb')

# MySQL
engine = create_engine('mysql+pymysql://user:password@localhost:3306/mydb')

# SQLite (файл)
engine = create_engine('sqlite:///path/to/database.db')

# SQLite (в памяти — для тестов)
engine = create_engine('sqlite:///:memory:')

Модели: декларативный стиль

Декларативный стиль — основной способ определения моделей в SQLAlchemy ORM. Класс модели наследует Base и описывает таблицу через атрибуты.

from sqlalchemy import Boolean, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, 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), nullable=False)
    age: Mapped[int | None] = mapped_column(Integer)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

Основные типы колонок

Параметры mapped_column()

Связи: ForeignKey и relationship()

ForeignKey — ограничение на уровне базы данных. Указывает, что значение колонки ссылается на первичный ключ другой таблицы.

relationship() — конструкция ORM. Позволяет обращаться к связанным объектам через Python-атрибуты.

Один-ко-многим (1:N)

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

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    addresses: Mapped[list["Address"]] = relationship(back_populates="user")

class Address(Base):
    __tablename__ = 'addresses'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    user: Mapped["User"] = relationship(back_populates="addresses")

Один-к-одному (1:1)

profile: Mapped["Profile"] = relationship(back_populates="user", uselist=False)

Многие-ко-многим (M:N)

from sqlalchemy import Column, ForeignKey, Table
tags_assoc = Table('tags_assoc', Base.metadata,
    Column('user_id', Integer, ForeignKey('users.id')),
    Column('tag_id', Integer, ForeignKey('tags.id'))
)
class User(Base):
    tags: Mapped[list["Tag"]] = relationship(
        secondary=tags_assoc,
        back_populates="users",
    )

Параметры relationship()

Сессия: жизненный цикл

Сессия — рабочий интерфейс между объектами Python и базой данных. Управляет транзакциями и отслеживает изменения объектов.

Создание сессии

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine('sqlite:///my_database.db')
with Session(engine) as session:
    ...

Состояния объекта в сессии

CRUD операции

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

# Один объект
new_user = User(name="Alice", age=30)
session.add(new_user)
session.commit()

# Несколько объектов
session.add_all([
    User(name='Bob', age=22),
    User(name='David', age=27),
])
session.commit()

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

from sqlalchemy import select

# Все записи
users = session.scalars(select(User)).all()

# По первичному ключу
user = session.get(User, 1)

# С фильтром
users = session.scalars(
    select(User).where(User.name == "Alice")
).all()

Update — обновление

user = session.get(User, 1)
if user:
    user.age = 35
    session.commit()

Delete — удаление

user = session.get(User, 1)
if user:
    session.delete(user)
    session.commit()

Запросы

Методы извлечения данных

Фильтрация

from sqlalchemy import and_, or_, select

# Простое условие
session.scalars(select(User).where(User.age > 25)).all()

# Функции: like, between, in_
session.scalars(select(User).where(User.name.like('A%'))).all()
session.scalars(select(User).where(User.id.between(2, 4))).all()
session.scalars(select(User).where(User.name.in_(["Alice", "Bob"]))).all()

# Логические условия
session.scalars(select(User).where(and_(User.age > 20, User.age < 30))).all()
session.scalars(select(User).where(or_(User.age > 30, User.name == 'David'))).all()

Сортировка, агрегации, группировка

from sqlalchemy import desc, func, select

# Сортировка
session.scalars(select(User).order_by(User.age)).all()
session.scalars(select(User).order_by(desc(User.age))).all()

# Агрегации
total = session.scalar(select(func.count(User.id)))
avg_age = session.scalar(select(func.avg(User.age)))

# Группировка
session.execute(
    select(User.name, func.count(User.id)).group_by(User.name)
).all()

# having() — фильтр после group_by
session.execute(
    select(User.age, func.count(User.id).label('cnt'))
    .group_by(User.age)
    .having(func.count(User.id) > 1)
).all()

JOIN

from sqlalchemy import select

# Inner Join
session.scalars(select(User).join(Address)).all()

# Left Outer Join
session.scalars(select(User).outerjoin(Address)).all()

Подзапросы

avg_age_sub = select(func.avg(User.age).label('avg_age')).subquery()
users = session.scalars(
    select(User).where(User.age > avg_age_sub.c.avg_age)
).all()