Старт: типова проблема з сесіями
Розробники FastAPI часто стикаються з ситуацією: застосунок працює локально, але на продакшені через 10 хвилин — помилка SSL SYSCALL або BrokenPipeError. Причина — пул з'єднань містить мертві сокети. SQLAlchemy 2.0 з опцією pool_pre_ping вирішує це, але правильне налаштування — лише частина шляху. Без коректної конфігурації асинхронних сесій та міграцій ви ризикуєте отримати N+1 запити та помилки MissingGreenlet під навантаженням.
Ми налаштовуємо SQLAlchemy для Python веб-застосунків на FastAPI та Flask вже понад п'ять років. За цей час зібрали набір best practices, які гарантують стабільність навіть при 1500+ запитах на секунду. У цій статті розберемо ключові компоненти: від асинхронної сесії до автоматичних міграцій Alembic. SQLAlchemy 2.0 Documentation рекомендує саме такий підхід.
Наприклад, в одному з проєктів з піковим навантаженням 2000 RPS ми зіткнулися з TimeoutError через відсутність pool_pre_ping. Після впровадження цієї опції та збільшення пулу до 30 з'єднань час відгуку знизився на 40%. Такі результати можливі лише при коректному налаштуванні всього ланцюжка.
Як налаштувати асинхронну сесію для FastAPI?
Асинхронність — стандарт для сучасних Python-фреймворків. Використовуємо create_async_engine з asyncpg:
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
pool_pre_ping=True перевіряє з'єднання перед використанням — обов'язково для продакшену. Без нього мертві з'єднання викликають 500-ті помилки, особливо в хмарних середовищах з тривалими таймаутами. Додатково налаштовуємо pool_recycle на 3600 секунд для автоматичної заміни старих з'єднань.
Впроваджуємо сесію через dependency injection: створюємо залежність get_db, яка відкриває сесію, виконує commit або rollback. Це стандартний патерн для FastAPI.
Чому expire_on_commit=False критичний для async?
За замовчуванням після commit() SQLAlchemy завершує всі об'єкти. При зверненні до атрибутів в async-режимі це викликає MissingGreenlet. Відключаємо — об'єкти залишаються доступними без зайвого запиту. Це підвищує продуктивність та усуває масу дебаг-сесій.
Моделі та запити в стилі 2.0
Новий типізований API: Mapped + mapped_column замість старого Column. Приклад моделі користувача з відношенням:
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")
lazy="selectin" — безпечна стратегія для async: виконується окремий SELECT ... WHERE id IN (...), без MissingGreenlet. У порівнянні з joinedload не створює гігантських JOIN-ів, що дає приріст продуктивності до 30% на вибірках з великою кількістю зв'язків.
Запити:
from sqlalchemy import select
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 with db.begin_nested():. Це зручно для rollback окремих операцій без відкату всієї транзакції.
Налаштування Alembic для async:
Ініціалізація:
alembic init -t async alembic
Правимо alembic/env.py:
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 # 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, 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()
compare_type=True — Alembic буде відстежувати зміни типів. Це економить час при рефакторингу.
Які помилки виникають при неправильному налаштуванні?
- MissingGreenlet — при лінивому завантаженні в async. Рішення: використовуйте lazy='selectin' або await db.refresh().
- N+1 queries — в async особливо небезпечні. Використовуйте selectinload або joinedload.
- Таймаути з'єднань — вирішуються через pool_pre_ping та pool_recycle.
- Гонка даних — транзакції повинні бути ідемпотентними. Наші інженери перевіряють це на етапі code review.
Порівняння синхронного та асинхронного підходів
| Критерій | Синхронний | Асинхронний |
|---|---|---|
| Драйвер | psycopg2 | asyncpg |
| Engine | create_engine | create_async_engine |
| Сесія | sessionmaker | async_sessionmaker |
| Запити | session.execute | await db.execute |
| Пропускна здатність | ~500 req/s | ~1500 req/s |
Асинхронний підхід дає приріст у 3 рази за кількістю запитів на секунду, що критично для високонавантажених проєктів.
Що входить в роботу
- Аудит поточної конфігурації SQLAlchemy та виявлення вузьких місць.
- Налаштування асинхронної сесії з pool_pre_ping, оптимізація пулу з'єднань.
- Проектування моделей з правильними lazy-стратегіями та типізацією.
- Реалізація міграцій Alembic з автогенерацією та контролем типів.
- Інтеграція сесії в FastAPI/Flask через dependency injection.
- Документація по експлуатації та інструкція по деплою.
- Підтримка після впровадження: 2 тижні консультацій.
Строки та вартість
Налаштування SQLAlchemy з нуля під новий проєкт — від 1 робочого дня. Міграція існуючого застосунку з 1.4 на 2.0 — від 2 днів. Вартість розраховується індивідуально після оцінки обсягу моделей та запитів.
Отримайте консультацію по вашому проєкту — наші спеціалісти допоможуть налаштувати SQLAlchemy так, щоб уникнути проблем під навантаженням. Замовте аудит поточної конфігурації та отримайте конкретні рекомендації щодо покращення продуктивності.