Как реализовать транзакции SQLAlchemy 2.0 в маршрутах FastAPI, сохраняя изоляцию тестов с помощью точек сохранения? ⇐ Python

[Anonymous]

Я использую FastAPI с SQLAlchemy 2.0 (Async). Я хочу следовать передовому опыту по заключению логики маршрута в транзакции. Однако я изо всех сил пытаюсь заставить это работать с моим пакетом Pytest, где я использую одну общую транзакцию для каждого теста, чтобы обеспечить легкий откат.
Путь:

Код: Выделить всё

@router.delete("/{id}")
async def delete_item(id: int, db: AsyncSession = Depends(get_db)):
async with db.begin(): # This is where the conflict happens
item = await db.get(Item, id)
await db.delete(item)

Тестовое приспособление:

Код: Выделить всё

engine: AsyncEngine = create_async_engine(
f"{SETTINGS.postgres.dsn}_test"
)

@pytest.fixture
async def db_session():
async with engine.connect() as connection:
async with connection.begin() as transaction:
async_session = AsyncSession(
bind=connection,
expire_on_commit=False,
)
yield async_session
await transaction.rollback()

Я использую pytest-asyncio с asyncio_mode = auto. Я подтвердил, что переопределение зависимостей работает, поэтому конфликт возникает исключительно на уровне состояния транзакции SQLAlchemy.
Конкретная ошибка:
При запуске тестов я получаю сообщение sqlalchemy.exc.InvalidRequestError: транзакция уже начата в этом сеансе. что ожидаемо, потому что во время тестов приспособление создает транзакцию верхнего уровня, и на маршруте мы пытаемся создать новую транзакцию. Однако я не хочу удалять транзакции из своей бизнес-логики только для того, чтобы выполнить тесты.
Мой вопрос:

• Каков рекомендуемый способ структурирования асинхронных транзакций SQLAlchemy 2.0, чтобы маршруты FastAPI могли использовать db.begin(), пока тесты выполняются внутри транзакции-обертки для отката?

Подробнее: https://stackoverflow.com/questions/798 ... aintaining