Настройка SQLAlchemy для Python веб-приложения

Настройка SQLAlchemy для Python веб-приложения

SQLAlchemy — стандарт де-факто для работы с реляционными БД в Python. В версии 2.0 синтаксис запросов кардинально изменился: устаревший session.query(Model) заменён на select(Model). Настройка с нуля подразумевает версию 2.x.

pip install sqlalchemy[asyncio] asyncpg alembic
# для синхронного варианта:
pip install sqlalchemy psycopg2-binary alembic

Engine и сессия

Для FastAPI и других async-фреймворков используем асинхронный engine:

# app/database.py
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/mydb"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,          # проверяет соединение перед использованием
    echo=False,                   # True — выводит SQL в stdout
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,       # объекты доступны после commit без re-fetch
)


class Base(DeclarativeBase):
    pass

pool_pre_ping=True критично для production: без него соединения, разорванные прокси или фаерволом по таймауту, возвращают ошибку вместо прозрачного переподключения.

Dependency для FastAPI

# app/deps.py
from collections.abc import AsyncGenerator
from app.database import AsyncSessionLocal
from sqlalchemy.ext.asyncio import AsyncSession


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Модели

# app/models/user.py
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Enum, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
import enum


class UserRole(enum.Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole), default=UserRole.viewer, nullable=False
    )
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now(), nullable=False
    )

    posts: Mapped[list["Post"]] = relationship(
        back_populates="author", lazy="selectin"
    )

Mapped с mapped_column — это новый типизированный API 2.0. Старый Column работает, но не даёт вывода типов в IDE.

lazy="selectin" для отношений в async-контексте — безопасная стратегия: SQLAlchemy выполнит отдельный SELECT ... WHERE id IN (...) вместо попытки ленивой загрузки, которая в async-режиме выбрасывает MissingGreenlet.

Запросы в стиле 2.0

from sqlalchemy import select, and_
from app.models.user import User
from app.models.post import Post


async def get_published_posts_with_authors(
    db: AsyncSession,
    limit: int = 20,
    offset: int = 0,
) -> list[Post]:
    stmt = (
        select(Post)
        .join(Post.author)
        .where(Post.status == "published")
        .order_by(Post.created_at.desc())
        .limit(limit)
        .offset(offset)
    )
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    stmt = select(User).where(User.email == email)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Транзакции и savepoint

async def transfer_ownership(
    db: AsyncSession,
    post_id: int,
    new_author_id: int,
) -> None:
    # Вложенная транзакция через savepoint
    async with db.begin_nested():
        post = await db.get(Post, post_id)
        if post is None:
            raise ValueError(f"Post {post_id} not found")
        post.author_id = new_author_id
    # Внешняя транзакция коммитится отдельно

Alembic: настройка миграций

alembic init -t async alembic

Правим alembic/env.py для async:

from logging.config import fileConfig
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
from app.database import Base
# импортируем все модели, чтобы Base.metadata их видел
import app.models  # noqa: F401

config = context.config
fileConfig(config.config_file_name)
target_metadata = Base.metadata


def run_migrations_online():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )

    async def do_run():
        async with connectable.connect() as connection:
            await connection.run_sync(
                context.configure,
                connection=connection,  # type: ignore[arg-type]
                target_metadata=target_metadata,
                compare_type=True,
            )
            async with context.begin_transaction():
                await connection.run_sync(context.run_migrations)

    import asyncio
    asyncio.run(do_run())


run_migrations_online()

alembic revision --autogenerate -m "create users"
alembic upgrade head

compare_type=True — Alembic будет замечать изменения типов колонок, не только добавление/удаление.

Сроки

Начальная настройка под FastAPI-проект: 1 день. Перевод существующего проекта с версии 1.4 (legacy Query API) на 2.0: 1–3 дня в зависимости от объёма моделей и нестандартных конструкций.