Как интегрировать SQLAlchemy с FastAPI для работы с базой данных?

Интеграция SQLAlchemy (async) с FastAPI

FastAPI полностью поддерживает асинхронный SQLAlchemy через asyncpg (PostgreSQL) или aiosqlite (SQLite). Схема работы: один AsyncEngine на всё приложение, фабрика сессий async_sessionmaker, и dependency injection через Depends.

1. Настройка engine и sessionmaker

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

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

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

AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,  # важно для async: объекты не перезапрашиваются
    class_=AsyncSession,
)

class Base(DeclarativeBase):
    pass

2. ORM-модели

# models.py
from sqlalchemy import String, Integer, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
from database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

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

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped[User] = relationship(back_populates="posts")

3. Dependency для сессии

# deps.py
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from 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

4. Router с CRUD-операциями

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Annotated
from deps import get_db
from models import User
from pydantic import BaseModel

router = APIRouter(prefix="/users", tags=["users"])

class UserOut(BaseModel):
    id: int
    username: str
    email: str
    model_config = {"from_attributes": True}

@router.get("/{user_id}", response_model=UserOut)
async def get_user(
    user_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    return user

@router.get("/", response_model=list[UserOut])
async def list_users(
    db: Annotated[AsyncSession, Depends(get_db)],
) -> list[User]:
    result = await db.scalars(select(User).order_by(User.id))
    return result.all()

5. Alembic для миграций

pip install alembic
alembic init alembic
# в alembic/env.py:
# from database import Base
# target_metadata = Base.metadata
alembic revision --autogenerate -m "create users"
alembic upgrade head

Подводные камни

• expire_on_commit=False обязателен в async-режиме: без него обращение к атрибутам объекта после коммита вызовет MissingGreenlet — сессия уже закрыта.
• Lazy loading (relationship(lazy="select")) не работает в async-контексте — используйте selectinload() или joinedload() из sqlalchemy.orm.
• db.get(User, id) использует кеш сессии (identity map) — для свежих данных из БД используйте select(User).where(User.id == id) с execution_options(populate_existing=True).
• Один engine на приложение — не создавайте engine внутри get_db(), иначе каждый запрос будет открывать новый пул соединений.
• При unit-тестах используйте отдельную тестовую БД и async_sessionmaker с NullPool для предотвращения утечек соединений между тестами.
• pool_pre_ping=True добавляет SELECT 1 перед каждым использованием соединения — полезно при долгих idle-паузах, но добавляет задержку.
• При работе с Mapped-аннотациями (SQLAlchemy 2.0) нельзя смешивать со старым стилем Column() в одной модели — выберите один подход.
• Alembic autogenerate не видит изменения в кастомных типах (например, pgvector) — добавляйте их вручную в revision-файл.

• Описывать sqlalchemy integration только как термин и не показывать механизм на минимальном примере.
• Игнорировать ошибки, пустые данные, конкурентный доступ или границы транзакции.
• Не связывать поведение с официальным контрактом FastAPI и реальной эксплуатацией.

• Объясняет sqlalchemy integration через последовательность действий, а не через набор ключевых слов.
• Приводит короткий кодовый пример или production-сценарий с ожидаемым поведением.
• Называет хотя бы один риск: производительность, безопасность, транзакции, память или сопровождение.
• Умеет обсудить отказ, наблюдаемость и rollback без изменения публичного контракта.