Работа с базами данных: SQLAlchemy + FastAPI

Привет! SQLAlchemy — это самый популярный ORM (Object-Relational Mapper) для Python. Он позволяет работать с базами данных через Python-объекты, а не писать сырые SQL-запросы. В сочетании с FastAPI это даёт мощный и удобный стек для разработки.

В этой статье мы разберём:

Настройку SQLAlchemy в FastAPI
Создание моделей
Управление сессиями
CRUD операции
Миграции с Alembic

Содержание

Что нужно знать перед началом
Основная часть
Установка зависимостей
Настройка подключения
Создание моделей
Создание Pydantic схем
CRUD операции
Эндпоинты FastAPI
Миграции с Alembic
Переменные окружения (.env)
Задачи для закрепления
Нюансы и подводные камни
Сессии и зависимость get_db()
N+1 проблема
Асинхронность
Частые ошибки и как их избежать
Ошибка 1: Забыл закрыть сессию
Ошибка 2: Неправильный импорт моделей
Ошибка 3: Конфликт миграций
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание работы с базами данных

Совет: SQLAlchemy — это ORM, который позволяет работать с БД через Python-объекты.

Основная часть

Установка зависимостей

pip install sqlalchemy databases asyncpg psycopg2-binary alembic python-dotenv

sqlalchemy — ORM
databases — асинхронный драйвер
asyncpg — драйвер для PostgreSQL
psycopg2-binary — драйвер для PostgreSQL (синхронный)
alembic — миграции
python-dotenv — переменные окружения

Совет: Для SQLite используй sqlite вместо asyncpg и psycopg2.

Настройка подключения

database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

load_dotenv()

# Настройка базы данных
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./app.db")

# Для SQLite нужно добавить connect_args
if DATABASE_URL.startswith("sqlite"):
    engine = create_engine(
        DATABASE_URL,
        connect_args={"check_same_thread": False}
    )
else:
    engine = create_engine(DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Зависимость для получения сессии
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Создание моделей

models.py:

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, Float, Text
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # Связи
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    published = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    author = relationship("User", back_populates="posts")

Совет: server_default=func.now() устанавливает текущее время на стороне БД.

Создание Pydantic схем

schemas.py:

from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional, List

# User схемы
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr

class UserCreate(UserBase):
    password: str = Field(..., min_length=6)

class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime
    updated_at: Optional[datetime]

    class Config:
        from_attributes = True

class UserWithPosts(UserResponse):
    posts: List['PostResponse'] = []

# Post схемы
class PostBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: Optional[str] = None
    published: bool = False

class PostCreate(PostBase):
    author_id: int

class PostResponse(PostBase):
    id: int
    author_id: int
    created_at: datetime
    updated_at: Optional[datetime]
    author: Optional[UserResponse]

    class Config:
        from_attributes = True

CRUD операции

crud.py:

from sqlalchemy.orm import Session
from models import User, Post
from schemas import UserCreate, PostCreate
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

# User CRUD
def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def get_user_by_username(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()

def get_user_by_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, user: UserCreate):
    hashed_password = get_password_hash(user.password)
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id: int, username: str = None, email: str = None):
    db_user = get_user(db, user_id)
    if not db_user:
        return None
    if username:
        db_user.username = username
    if email:
        db_user.email = email
    db.commit()
    db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    db_user = get_user(db, user_id)
    if not db_user:
        return False
    db.delete(db_user)
    db.commit()
    return True

# Post CRUD
def get_post(db: Session, post_id: int):
    return db.query(Post).filter(Post.id == post_id).first()

def get_posts(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Post).offset(skip).limit(limit).all()

def get_posts_by_user(db: Session, user_id: int, skip: int = 0, limit: int = 10):
    return db.query(Post).filter(Post.author_id == user_id).offset(skip).limit(limit).all()

def create_post(db: Session, post: PostCreate):
    db_post = Post(
        title=post.title,
        content=post.content,
        published=post.published,
        author_id=post.author_id
    )
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

def update_post(db: Session, post_id: int, title: str = None, content: str = None, published: bool = None):
    db_post = get_post(db, post_id)
    if not db_post:
        return None
    if title is not None:
        db_post.title = title
    if content is not None:
        db_post.content = content
    if published is not None:
        db_post.published = published
    db.commit()
    db.refresh(db_post)
    return db_post

def delete_post(db: Session, post_id: int):
    db_post = get_post(db, post_id)
    if not db_post:
        return False
    db.delete(db_post)
    db.commit()
    return True

Эндпоинты FastAPI

main.py:

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from database import get_db
from models import Base, engine
import crud
import schemas

# Создаём таблицы
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Blog API", version="1.0.0")

# ======== USERS ========

@app.get("/users", response_model=List[schemas.UserResponse])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.get("/users/{user_id}", response_model=schemas.UserWithPosts)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.post("/users", response_model=schemas.UserResponse, status_code=201)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    if crud.get_user_by_username(db, user.username):
        raise HTTPException(status_code=400, detail="Username already exists")
    if crud.get_user_by_email(db, user.email):
        raise HTTPException(status_code=400, detail="Email already exists")
    return crud.create_user(db, user)

@app.put("/users/{user_id}", response_model=schemas.UserResponse)
def update_user(
    user_id: int,
    user_update: schemas.UserUpdate,
    db: Session = Depends(get_db)
):
    updated = crud.update_user(db, user_id, user_update.username, user_update.email)
    if not updated:
        raise HTTPException(status_code=404, detail="User not found")
    return updated

@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    if not crud.delete_user(db, user_id):
        raise HTTPException(status_code=404, detail="User not found")
    return None

# ======== POSTS ========

@app.get("/posts", response_model=List[schemas.PostResponse])
def read_posts(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    posts = crud.get_posts(db, skip=skip, limit=limit)
    return posts

@app.get("/posts/{post_id}", response_model=schemas.PostResponse)
def read_post(post_id: int, db: Session = Depends(get_db)):
    post = crud.get_post(db, post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

@app.post("/posts", response_model=schemas.PostResponse, status_code=201)
def create_post(post: schemas.PostCreate, db: Session = Depends(get_db)):
    if not crud.get_user(db, post.author_id):
        raise HTTPException(status_code=404, detail="Author not found")
    return crud.create_post(db, post)

@app.put("/posts/{post_id}", response_model=schemas.PostResponse)
def update_post(
    post_id: int,
    post_update: schemas.PostUpdate,
    db: Session = Depends(get_db)
):
    updated = crud.update_post(
        db, post_id,
        post_update.title,
        post_update.content,
        post_update.published
    )
    if not updated:
        raise HTTPException(status_code=404, detail="Post not found")
    return updated

@app.delete("/posts/{post_id}", status_code=204)
def delete_post(post_id: int, db: Session = Depends(get_db)):
    if not crud.delete_post(db, post_id):
        raise HTTPException(status_code=404, detail="Post not found")
    return None

@app.get("/users/{user_id}/posts", response_model=List[schemas.PostResponse])
def read_user_posts(user_id: int, skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    if not crud.get_user(db, user_id):
        raise HTTPException(status_code=404, detail="User not found")
    posts = crud.get_posts_by_user(db, user_id, skip=skip, limit=limit)
    return posts

Миграции с Alembic

Инициализация Alembic:

alembic init alembic

Настройка alembic.ini:

sqlalchemy.url = postgresql://user:password@localhost/dbname

Настройка env.py:

from models import Base
target_metadata = Base.metadata

Создание миграции:

alembic revision --autogenerate -m "Initial migration"

Применение миграции:

alembic upgrade head

Совет: Alembic автоматически генерирует миграции на основе изменений в моделях.

Переменные окружения (`.env`)

DATABASE_URL=postgresql://user:password@localhost:5432/dbname
# или для SQLite
DATABASE_URL=sqlite:///./app.db
SECRET_KEY=your-secret-key
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

Задачи для закрепления

Задача 1. Создай модель Category с полями id, name, description.

Задача 2. Добавь связь Post → Category (многие к одному).

Задача 3. Напиши CRUD для категорий.

Задача 4. Добавь эндпоинты для категорий.

Задача 5. Создай миграцию для добавления поля category_id.

Ответы:

Задача 1.

class Category(Base):
    __tablename__ = "categories"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False, unique=True)
    description = Column(Text)

Задача 2.

# В модели Post добавляем:
category_id = Column(Integer, ForeignKey("categories.id"))
category = relationship("Category", back_populates="posts")

# В модели Category добавляем:
posts = relationship("Post", back_populates="category")

Задача 3.

def get_category(db: Session, category_id: int):
    return db.query(Category).filter(Category.id == category_id).first()

def get_categories(db: Session, skip=0, limit=10):
    return db.query(Category).offset(skip).limit(limit).all()

def create_category(db: Session, name: str, description: str = None):
    db_category = Category(name=name, description=description)
    db.add(db_category)
    db.commit()
    db.refresh(db_category)
    return db_category

Задача 4.

@app.get("/categories", response_model=List[schemas.CategoryResponse])
def get_categories(skip=0, limit=10, db: Session = Depends(get_db)):
    return crud.get_categories(db, skip, limit)

@app.post("/categories", response_model=schemas.CategoryResponse)
def create_category(category: schemas.CategoryCreate, db: Session = Depends(get_db)):
    return crud.create_category(db, category.name, category.description)

Задача 5.

alembic revision --autogenerate -m "Add category_id to posts"
alembic upgrade head

Нюансы и подводные камни

Сессии и зависимость `get_db()`

Сессия создаётся для каждого запроса и закрывается после. Используй yield в зависимости.

N+1 проблема

При запросе списка постов с авторами может возникнуть N+1 запросов. Используй joinedload() для подгрузки связанных данных.

from sqlalchemy.orm import joinedload

def get_posts_with_users(db: Session):
    return db.query(Post).options(joinedload(Post.author)).all()

Асинхронность

Для асинхронной работы используй asyncpg и databases.

Частые ошибки и как их избежать

Ошибка 1: Забыл закрыть сессию

Используй finally в get_db() или менеджер контекста.

Ошибка 2: Неправильный импорт моделей

Убедись, что модели импортированы перед Base.metadata.create_all().

Ошибка 3: Конфликт миграций

Перед созданием миграции убедись, что модели синхронизированы с базой.

Шпаргалка

Компонент	Назначение
`SessionLocal`	Фабрика сессий
`Base`	Базовый класс для моделей
`get_db()`	Зависимость для получения сессии
`@app.on_event("startup")`	Создание таблиц при старте
`alembic`	Управление миграциями

Заключение

Сегодня мы:

Настроили SQLAlchemy в FastAPI
Создали модели и Pydantic схемы
Реализовали CRUD операции
Настроили миграции с Alembic

КВИЗ

Что дальше?

Следующая статья: «Аутентификация JWT в FastAPI»