</>
halfcoder

Гайды

SQLAlchemy: работа с БД в Python

DeclarativeBase 2.0, select и сессии, relationship и загрузчики, async с asyncpg, Core и транзакции.

~12 мин чтения

BackendPythonSQLAlchemyORM

SQLAlchemy: работа с БД в Python

SQLAlchemy 2.0 — единый стиль Core (SQL-выражения) и ORM (модели, сессии). Асинхронный стек с asyncpg хорошо сочетается с FastAPI — см. Асинхронные эндпоинты и asyncpg. Миграции — Alembic; Django ORM — оптимизация запросов в Django.

1. Движок и сессия (sync)

python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

class Base(DeclarativeBase):
    pass

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

def get_session():
    s = Session()
    try:
        yield s
    finally:
        s.close()

Для FastAPI оборачивают в Depends — см. asyncpg и официальный раздел.

2. Модель

python
from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = "users"

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

Mapped[] + mapped_column — стиль 2.0.

3. Запросы ORM

python
from sqlalchemy import select

with Session() as session:
    u = session.scalars(select(User).where(User.email == "a@b.c")).one_or_none()
    session.add(User(email="x@y.z", name="Ann"))
    session.commit()

select() из sqlalchemy.

4. Отношения

python
from __future__ import annotations
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, selectinload

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255))
    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"))
    user: Mapped[User] = relationship(back_populates="orders")

selectinload, joinedload — аналоги prefetch_related / select_related.

python
select(User).options(selectinload(User.orders))

5. Async (кратко)

python
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

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

async with AsyncSessionLocal() as session:
    await session.execute(select(User).where(User.id == 1))
    await session.commit()

6. Core: текст и безопасность

python
from sqlalchemy import text
session.execute(text("SELECT 1 WHERE id = :id"), {"id": x})

Параметры только через bind, не через f-string.

7. Транзакции

python
with Session() as session:
    with session.begin():
        session.add(...)
        session.execute(...)

При исключении — откат.

8. Обновления и конфликты

session.merge(obj) удобен при десериализации DTO с частично известным PK; помните про отдельный SELECT и гонки — для высокой конкуренции чаще UPDATE ... WHERE через Core или версионирование строки (xmin/version_id). Удаление каскадом настраивается в relationship(..., cascade="all, delete-orphan") — проверьте, что не рвёте связи случайно в админских скриптах.

9. После commit: expire_on_commit

По умолчанию после commit() атрибуты экземпляра ORM помечаются как expired; при следующем обращении к полю SQLAlchemy может выполнить implicit lazy load — если сессия уже закрыта, получите ошибку или N+1 в неожиданном месте. Для ответов API либо sessionmaker(..., expire_on_commit=False) (осознанно: «устаревшие» данные в памяти до явного refresh), либо model_validate в Pydantic / dict до закрытия сессии.

10. Чек-лист

  • Один стиль 2.0 (select, Mapped).
  • Явные relationship и стратегии загрузки для списков.
  • Пул и pool_pre_ping=True для продакшена.
  • Политика expire_on_commit согласована с тем, как вы отдаёте объекты из эндпоинтов.
  • Alembic для изменения схемы — гайд.

Дальше: Flask + SQLAlchemy · тег SQLAlchemy