Какую ORM используешь в FastAPI?

ORM в FastAPI

В современных проектах на FastAPI наиболее популярна и рекомендуется SQLAlchemy 2.0+ с асинхронным драйвером. В нашем проекте также используется подход с Goose для миграций, что даёт полный контроль над схемой базы данных.

SQLAlchemy 2.0 в FastAPI

SQLAlchemy 2.0 — это выбор для production-приложений. Она обеспечивает типизацию, отличное удобство разработки и производительность.

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime
from typing import AsyncGenerator

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

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    future=True,
    pool_size=20,
    max_overflow=0,
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
)

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    username = Column(String(255), unique=True, index=True)
    email = Column(String(255), unique=True, index=True)
    created_at = Column(DateTime(timezone=True), default=datetime.utcnow)

Dependency Injection для сессий в FastAPI

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

@app.post("/users")
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    user = User(**user_data.dict())
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

Репозиторий паттерн

Для чистой архитектуры часто используется паттерн Repository, который отделяет логику доступа к данным от бизнес-логики:

from typing import Optional, List
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def get_by_id(self, user_id: int) -> Optional[User]:
        stmt = select(User).where(User.id == user_id)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
        stmt = select(User).offset(skip).limit(limit)
        result = await self.session.execute(stmt)
        return result.scalars().all()
    
    async def create(self, user: User) -> User:
        self.session.add(user)
        await self.session.commit()
        await self.session.refresh(user)
        return user
    
    async def update(self, user_id: int, **kwargs) -> Optional[User]:
        user = await self.get_by_id(user_id)
        if user:
            for key, value in kwargs.items():
                setattr(user, key, value)
            await self.session.commit()
            await self.session.refresh(user)
        return user
    
    async def delete(self, user_id: int) -> bool:
        user = await self.get_by_id(user_id)
        if user:
            await self.session.delete(user)
            await self.session.commit()
            return True
        return False

Service слой

Business logic отделяется в Service слой, который использует Repository:

from pydantic import BaseModel

class UserCreate(BaseModel):
    username: str
    email: str

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo
    
    async def create_user(self, user_data: UserCreate) -> User:
        # Проверка уникальности
        existing = await self.repo.get_by_email(user_data.email)
        if existing:
            raise ValueError(f"User with email {user_data.email} already exists")
        
        user = User(
            username=user_data.username,
            email=user_data.email
        )
        return await self.repo.create(user)

Использование в handler'ах

from fastapi import APIRouter, Depends, HTTPException

router = APIRouter(prefix="/api/v1/users")

async def get_user_service(db: AsyncSession = Depends(get_db)) -> UserService:
    repo = UserRepository(db)
    return UserService(repo)

@router.post("/users")
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service)
):
    try:
        user = await service.create_user(user_data)
        return {"id": user.id, "username": user.username, "email": user.email}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/users/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    user = await service.repo.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Миграции с Goose

Вместо Alembic используется Goose для raw SQL миграций, что обеспечивает полный контроль и совместимость с любым ORM:

-- migrations/00001_create_users_table.sql
-- +goose Up
CREATE TABLE users (
    id BIGSERIAL PRIMARY KEY,
    username VARCHAR(255) UNIQUE NOT NULL,
    email VARCHAR(255) UNIQUE NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- +goose Down
DROP TABLE users;

Async/await паттерны

# ❌ Плохо - блокирует
users = session.query(User).all()

# ✅ Хорошо - асинхронно
stmt = select(User)
result = await session.execute(stmt)
users = result.scalars().all()

# ✅ Для сложных запросов с joins
stmt = select(User).join(Post).where(Post.published == True)
result = await session.execute(stmt)
users = result.unique().scalars().all()

Лучшие практики

• Используй async/await: не блокируй event loop
• Отделяй Repository от Service: чистая архитектура
• Используй type hints: полная типизация
• Избегай N+1 запросов: используй selectinload для relationship'ов
• Раннее закрытие сессий: используй async context managers
• Raw SQL для сложного: SQL сложнее писать, чем ORM для простых операций

Сочетание SQLAlchemy + Goose + Repository паттерна дает надежную, масштабируемую и хорошо протестируемую архитектуру.