Как организовать async-работу с базой данных (async SQLAlchemy, asyncpg)?

Когда нужен async-доступ к БД

Асинхронная база полезна, если приложение уже построено на async I/O и одновременно ожидает множество операций с внешними сервисами. Она не ускоряет сам SQL-запрос и не заменяет индексы или оптимизацию схемы; выигрыш появляется за счёт неблокирующего ожидания I/O.

Создание AsyncEngine

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

DATABASE_URL = 'postgresql+asyncpg://user:password@db:5432/app_db'

engine = create_async_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=10,
)
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Dependency сессии

from collections.abc import AsyncGenerator


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

Запрос через AsyncSession

from fastapi import Depends
from sqlalchemy import select


@app.get('/articles/{article_id}')
async def get_article(article_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Article).where(Article.id == article_id)
    )
    article = result.scalar_one_or_none()
    if article is None:
        raise HTTPException(status_code=404, detail='Article not found')
    return article

Транзакции

Транзакционные сценарии лучше размещать в service layer и использовать явные границы через async with session.begin(). Это позволяет безопасно объединить несколько операций и автоматически выполнить commit при успехе либо rollback при исключении.

async with db.begin():
    db.add(order)
    db.add(payment)

Пулы и ограничения

Размер pool нужно считать на уровне всей инфраструктуры: число реплик × workers × pool size не должно превышать лимит PostgreSQL. Нельзя создавать новый engine или session factory на каждый запрос. При высокой нагрузке нужны тайм-ауты, мониторинг пула, защита от утечек соединений и оптимизация медленных запросов.

Как ответить на собеседовании

Для async PostgreSQL использую SQLAlchemy AsyncEngine с драйвером asyncpg, создаю один engine на процесс и выдаю AsyncSession на запрос через dependency. Запросы выполняю с await, транзакции ограничиваю сервисным слоем, а размер пула выбираю с учётом всех workers и лимита базы.