Что такое SQLAlchemy и как использовать его асинхронный режим?

SQLAlchemy — это мощный инструментарий и ORM для Python, который позволяет работать с реляционными базами данных, абстрагируя SQL-запросы в Python-объекты. Асинхронный режим, добавленный в версии 1.4, позволяет интегрировать SQLAlchemy с асинхронными фреймворками, такими как FastAPI или aiohttp, обеспечивая неблокирующие операции с БД.

Основные компоненты асинхронного режима

• create_async_engine: Создаёт асинхронный движок для подключения к БД. Требует асинхронного драйвера (например, asyncpg для PostgreSQL, aiomysql для MySQL).
• AsyncSession: Сессия для управления асинхронными транзакциями и операциями.
• AsyncConnection: Прямое асинхронное подключение для выполнения сырых SQL-запросов.

Пример настройки и использования

Сначала установите зависимости и настройте движок. Затем определите модели (они такие же, как в синхронном режиме). Далее используйте асинхронный контекстный менеджер для сессии.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy import Column, Integer, String

# Создаём асинхронный движок (пример для PostgreSQL с asyncpg)
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
async_engine = create_async_engine(DATABASE_URL, echo=True)

# Базовый класс для моделей
Base = declarative_base()

# Определяем модель
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

# Создаём фабрику асинхронных сессий
AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

# Асинхронная функция для работы с БД
async def create_user(user_name: str):
    async with AsyncSessionLocal() as session:
        new_user = User(name=user_name)
        session.add(new_user)
        await session.commit()
        await session.refresh(new_user)
        return new_user

# Пример запроса с фильтром
async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            select(User).where(User.name.like("A%"))
        )
        users = result.scalars().all()
        return users

Где и как применять

Асинхронный режим SQLAlchemy идеально подходит для веб-приложений на FastAPI, aiohttp или любых asyncio-проектах, где требуется высокая производительность и масштабируемость. Он позволяет обрабатывать тысячи одновременных подключений к БД без создания отдельных потоков. Важно помнить, что не все расширения SQLAlchemy поддерживают асинхронный режим, и для сложных операций может потребоваться использование синхронных методов в отдельных потоках.

Вывод: Используйте асинхронный режим SQLAlchemy при разработке асинхронных веб-приложений на Python, где критична производительность ввода-вывода и необходимо эффективно управлять множеством одновременных запросов к базе данных.