SQLAlchemy ORM: модели, session, queries

SQLAlchemy - самая используемая ORM в Python. С версии 2.0 (2023) API сильно обновлён: typed, более идиоматичный, async support из коробки. В этом уроке - модели через DeclarativeBase, сессии, базовые запросы и отношения (под капотом это всё те же JOIN). Используем SQLAlchemy 2.0 API. Предполагается, что SQL ты уже видел - если нет, начни с трека SQL.

Установка и подключение

pip install sqlalchemy psycopg[binary]   # для PostgreSQL
pip install sqlalchemy aiosqlite          # для async SQLite

Подключение:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg://user:pass@localhost/dbname", echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

echo=True логирует все SQL - полезно в разработке. sessionmaker фабрика сессий. На production обычно echo=False.

Модели через DeclarativeBase

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

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)

    orders: Mapped[list["Order"]] = relationship(back_populates="user")

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    total: Mapped[float]

    user: Mapped["User"] = relationship(back_populates="orders")

С 2.0 используется Mapped[type] и mapped_column() - типизированный API, mypy friendly. Старый стиль (Column(Integer)) тоже работает для backward compat.

Создание таблиц

Base.metadata.create_all(engine)

Создаст все таблицы определённые в Base. В production используется Alembic для миграций (урок 52).

Session - unit of work

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Создание
    user = User(name="Alice", email="alice@example.com")
    session.add(user)
    session.commit()
    print(user.id)   # после commit получит ID

Session это unit of work - track изменений и flush в БД одним коммитом. add() ставит в очередь, commit() отправляет в БД и подтверждает транзакцию.

С context manager автоматический rollback при exception.

SELECT через select()

from sqlalchemy import select

with Session(engine) as session:
    # Один по PK
    user = session.get(User, 1)

    # Поиск
    stmt = select(User).where(User.email == "alice@example.com")
    user = session.scalars(stmt).first()
    # или session.scalar(stmt) для одного

    # Список
    stmt = select(User).where(User.name.like("A%")).order_by(User.id)
    users = session.scalars(stmt).all()

    # С limit/offset
    stmt = select(User).order_by(User.id).limit(10).offset(20)
    users = session.scalars(stmt).all()

select(Model) - новый стиль (vs старый session.query(Model)). Возвращает Select, который запускается через session методы.

UPDATE

with Session(engine) as session:
    user = session.get(User, 1)
    user.name = "Alice Smith"
    session.commit()
    # SQL: UPDATE users SET name = ? WHERE id = ?

Изменения tracked автоматически. На commit SQLAlchemy сгенерирует UPDATE.

Для bulk update:

from sqlalchemy import update

stmt = update(User).where(User.id.in_([1, 2, 3])).values(active=True)
session.execute(stmt)
session.commit()

DELETE

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

# Bulk
from sqlalchemy import delete
stmt = delete(User).where(User.active == False)
session.execute(stmt)
session.commit()

Relationships - один-ко-многим

Из примера выше: User имеет много Order, Order принадлежит одному User.

with Session(engine) as session:
    user = session.get(User, 1)
    print(user.orders)   # список Order - lazy loaded query

    # Создание связанного
    order = Order(total=100.0)
    user.orders.append(order)
    session.commit()
    # SQL автоматически: INSERT INTO orders ... с user_id=1

Eager loading - избегаем N+1

# N+1 ловушка
users = session.scalars(select(User)).all()   # 1 запрос
for user in users:
    print(user.orders)   # N запросов - по одному на каждого user!

# Solution - selectinload
from sqlalchemy.orm import selectinload

stmt = select(User).options(selectinload(User.orders))
users = session.scalars(stmt).all()   # 2 запроса (users + orders в одном запросе)

selectinload (для коллекций) и joinedload (для many-to-one) предзагружают связанные данные. Critical для performance.

Many-to-many

from sqlalchemy import Table, Column, ForeignKey

user_role = Table(
    "user_role",
    Base.metadata,
    Column("user_id", ForeignKey("users.id"), primary_key=True),
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
)

class User(Base):
    # ...
    roles: Mapped[list["Role"]] = relationship(secondary=user_role, back_populates="users")

class Role(Base):
    __tablename__ = "roles"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    users: Mapped[list["User"]] = relationship(secondary=user_role, back_populates="roles")

secondary указывает junction table. Удобно для связей вроде user-role, post-tag.

Aggregation

from sqlalchemy import func

stmt = select(func.count(User.id))
total = session.scalar(stmt)

stmt = select(User.id, func.sum(Order.total)).join(Order).group_by(User.id)
results = session.execute(stmt).all()   # list of tuples

func для SQL функций (count, sum, max, avg). group_by для агрегации.

Async SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://...")

async def get_user(id: int):
    async with AsyncSession(engine) as session:
        user = await session.get(User, id)
        return user

async def list_users():
    async with AsyncSession(engine) as session:
        stmt = select(User).where(User.active == True)
        result = await session.scalars(stmt)
        return result.all()

Async требует async driver (asyncpg для PostgreSQL, aiosqlite для SQLite). Использовать в FastAPI с async handlers - получишь полностью non-blocking стек.

FastAPI + SQLAlchemy

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession

engine = create_async_engine("postgresql+asyncpg://...")
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_session():
    async with SessionFactory() as session:
        yield session

app = FastAPI()

@app.get("/users/{id}")
async def get_user(id: int, session: AsyncSession = Depends(get_session)):
    user = await session.get(User, id)
    if user is None:
        raise HTTPException(404, "Not found")
    return user

Depends(get_session) гарантирует cleanup через yield-fixture. expire_on_commit=False важно для async, иначе attributes lazy-load после commit и нужны await что не работает после yield.

Transactions

with Session(engine) as session:
    try:
        user = User(name="Alice", email="alice@b.c")
        session.add(user)

        order = Order(user_id=user.id, total=100)   # нужно flush для user.id
        session.flush()   # SQL INSERT user, но без commit
        session.add(order)

        session.commit()   # одна транзакция: и user, и order
    except Exception:
        session.rollback()
        raise

flush() отправляет SQL без commit - получаем generated IDs. commit() финализирует. rollback() откатывает всё.

Context manager делает commit/rollback автоматически:

with Session(engine) as session:
    with session.begin():
        user = User(...)
        session.add(user)
        # commit при выходе из inner with, rollback на exception

Identity map

SQLAlchemy кеширует загруженные объекты в session:

with Session(engine) as session:
    user1 = session.get(User, 1)
    user2 = session.get(User, 1)
    print(user1 is user2)   # True - тот же объект, не два SELECT

Identity map гарантирует один объект на (session, type, PK). Это оптимизация и consistency.

Raw SQL когда нужно

from sqlalchemy import text

result = session.execute(text("SELECT * FROM users WHERE name = :name"), {"name": "Alice"})
for row in result:
    print(row)

Не злоупотребляй - теряются преимущества ORM. Но иногда complex queries проще на raw SQL.

Распространённые ошибки

1. N+1 query

Уже обсуждали. Используй eager loading.

2. Забыть commit

session.add(user)
# session.commit() - забыли
session.close()   # изменения не сохранены

Context manager помогает: commit явно нужен, но close через with гарантирован.

3. Mutable session между requests

# Плохо в FastAPI
session = SessionLocal()   # один на всю программу

@app.get("/users/{id}")
def get_user(id: int):
    return session.get(User, id)   # shared state - race conditions

Каждый request должен получить свою session. Через Depends с yield.

4. Detached object access

with Session(engine) as session:
    user = session.get(User, 1)
# session закрыта
print(user.orders)   # DetachedInstanceError - lazy load невозможен

Используй eager loading или работай в одной session.

5. Mixing sync и async моделей

Sync и async SQLAlchemy используют разные engines. Не путай. Если приложение async - везде AsyncSession.

SQLAlchemy + Pydantic

Часто разделяют ORM модели (для БД) и Pydantic модели (для API):

# ORM
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    password_hash: Mapped[str]

# Pydantic
class UserOut(BaseModel):
    id: int
    name: str
    # password_hash намеренно не включён

    model_config = ConfigDict(from_attributes=True)

# Conversion
user_orm = session.get(User, 1)
user_out = UserOut.model_validate(user_orm)

from_attributes=True позволяет Pydantic создаваться из объектов с атрибутами. Это standard pattern: ORM модели для persistence, Pydantic для API.

Сравнение с Go и PHP

В Go стандарт - database/sql (raw SQL) или sqlx (тонкая обёртка). ORM в Go менее популярны (GORM существует но не доминирует). Go разработчики чаще пишут SQL явно.

В PHP - Doctrine ORM наиболее популярный. Похож на SQLAlchemy: модели как классы, entity manager как session.

// Doctrine
$user = $entityManager->find(User::class, 1);
$user->setName('New Name');
$entityManager->flush();

SQLAlchemy более мощный, но и более сложный. Идиоматический Python backend часто строится на FastAPI + SQLAlchemy + Pydantic.

Мини-задание

  1. Модель + создание:
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author: Mapped[str]

engine = create_engine("sqlite:///books.db")
Base.metadata.create_all(engine)

with Session(engine) as session:
    book = Book(title="Test", author="Alice")
    session.add(book)
    session.commit()
    print(book.id)
  1. SELECT:
from sqlalchemy import select

with Session(engine) as session:
    stmt = select(Book).where(Book.author == "Alice").order_by(Book.id)
    books = session.scalars(stmt).all()
    for b in books:
        print(b.title)
  1. Async с FastAPI (упрощённо):
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine("sqlite+aiosqlite:///./books.db")
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_session():
    async with SessionFactory() as session:
        yield session

app = FastAPI()

@app.get("/books")
async def list_books(session: AsyncSession = Depends(get_session)):
    result = await session.scalars(select(Book))
    return [{"id": b.id, "title": b.title} for b in result.all()]

Что дальше

Освоили SQLAlchemy. ORM-модели не стоит отдавать наружу как есть - для ответов API описывай отдельные Pydantic-схемы. В следующем уроке - Alembic: миграции БД для управления schema changes в production. Это критический компонент любого приложения с БД.

Зарегистрируйтесь бесплатно, чтобы пройти квиз, решить задание с автопроверкой и вести прогресс.