SQLAlchemy в FastAPI: одна Session на запрос и явная транзакция

Маршрут сохранил объект, но следующий запрос его не видит. Или наоборот: после одной ошибки начинает падать весь API. Часто причина в границах Session: её создали глобально, забыли commit или рассчитывали, что закрытие автоматически сохранит изменения. Для небольшого FastAPI-сервиса полезно сначала выбрать простой и явный порядок работы с базой.

У каждого запроса своя Session

Engine создаёт подключения и управляет их пулом; он может жить на уровне приложения. Session хранит состояние ORM-операций и текущей транзакции. Её удобно выдавать обработчику через зависимость с yield и закрывать при завершении использования. Не делайте один глобальный экземпляр Session для всех клиентов: параллельные запросы не должны делить её изменяемое состояние.

Закрытие освобождает ресурсы, но не означает commit. В примере успешный маршрут сам сохраняет изменения. Если операция состоит из нескольких действий, объедините их в одну транзакцию на уровне сервиса. Чтение также может начать транзакцию автоматически, поэтому после работы Session нужно закрыть даже тогда, когда вы ничего не меняли.

Проверим сохранение между запросами

Для запуска нужны FastAPI, HTTPX и SQLAlchemy 2. Временная SQLite находится в памяти; StaticPool сохраняет одну базу для локальных HTTP-тестов. Здесь он выбран для тестового примера, а не как универсальная настройка production-сервера. Первый запрос создаёт запись, второй читает её в другой Session.

Сохраните весь блок в файл и запустите Python 3.11+ в отдельном окружении с пакетами fastapi, httpx, sqlalchemy. Для моделей используется Pydantic v2; отдельный сервер для TestClient не нужен.

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, String, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.pool import StaticPool

class Base(DeclarativeBase):
    pass
class Item(Base):
    __tablename__ = "items"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, unique=True)

engine = create_engine("sqlite://", poolclass=StaticPool,
                       connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
app = FastAPI()
def get_db():
    with Session(engine) as session:
        yield session
Db = Annotated[Session, Depends(get_db)]

@app.post("/items", status_code=201)
def create(name: str, db: Db):
    item = Item(name=name)
    db.add(item)
    db.commit()
    db.refresh(item)
    return {"id": item.id, "name": item.name}

@app.get("/items/{item_id}")
def read(item_id: int, db: Db):
    item = db.get(Item, item_id)
    if item is None:
        raise HTTPException(404, "item not found")
    return {"id": item.id, "name": item.name}

with TestClient(app) as client:
    created = client.post("/items", params={"name": "Desk"})
    assert created.status_code == 201
    item = created.json()
    assert client.get(f"/items/{item['id']}").json() == item
    assert client.get("/items/999").status_code == 404
print("committed item:", item)
engine.dispose()

Почему HTTP-тест сильнее проверки объекта

Сразу после db.add объект существует в Python, но это ещё не подтверждает запись в базу. Второй запрос создаёт другую Session и читает сохранённое состояние. Такой сценарий обнаруживает пропущенный commit. Если тест использует тот же объект Session для записи и чтения, identity map может скрыть ошибку границ транзакции.

refresh загружает текущее состояние объекта после commit, в том числе значения, назначенные базой. Возвращаем обычный словарь из скалярных полей, пока Session доступна. Для ответа с отношениями заранее определите, какие данные загрузить: попытка ленивого чтения после закрытия Session становится отдельным источником ошибок сериализации.

Где выполнять commit в большом проекте

Когда один запрос создаёт заказ и резервирует остаток, commit внутри каждого репозитория неудобен: первая половина может сохраниться до отказа второй. Репозитории добавляют и читают объекты, а сервис завершает общую транзакцию. В простом маршруте выше все действия видны рядом, поэтому ручной commit делает учебный договор очевидным.

Не прячьте ошибки сохранения в финализаторе зависимости после формирования результата. Клиенту нужен ответ, соответствующий факту выполнения операции. Выполняйте существенную запись там, где ещё можно обработать известный конфликт и выбрать статус HTTP. Закрытие зависимости оставьте для освобождения ресурса.

Следующая практика

В открытом уроке Engine, Session и SQLite вы подключите базу к учебному API. Сохраните привычку проверять запись следующим запросом: она помогает отличить живой ORM-объект от действительно сохранённой строки. Откройте практику Session и SQLite.

Чтобы собрать маршруты, хранение и тесты в одном приложении, откройте бесплатную программу FastAPI. Когда понадобится отдельная PostgreSQL и воспроизводимый запуск окружения, продолжите в Docker и Docker Compose: вы соедините сервис с базой, настроите сеть, том и проверку готовности.