Налаштування 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,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

class Base(DeclarativeBase):
    pass

Dependency для FastAPI

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

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 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"
    )

Запити у стилі 2.0

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

async def get_published_posts(
    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()

Транзакції

async def transfer_ownership(
    db: AsyncSession,
    post_id: int,
    new_author_id: int,
) -> None:
    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
import app.models

config = context.config
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,
                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()

Терміни

Первісна настройка під FastAPI-проект: 1 день. Перевід існуючого проекту з версії 1.4 (legacy Query API) на 2.0: 1–3 дні залежно від обсягу моделей.