Как работают транзакции с FastAPI Depends()? ⇐ Python

[Anonymous]

Я пытаюсь использовать FastAPI Depends() как способ уменьшить повторение (с помощью Piccolo ORM). Я не могу найти исходную страницу FastAPI, на которой был основан этот код, но она есть где-то в документации. SQLite требует другого типа транзакции для определенных комбинаций операторов SQL, и я бы предпочел не использовать оператор with в каждой отдельной функции.
Пример кода приведен ниже, но я хотел бы понять:

• Что здесь делают Depends() и выход на самом деле?
  • Это создает Транзакция IMMEDIATE правильно, или она работает не так, как я предполагал?
• Если у меня неправильный код, есть ли другой способ уменьшить повторение?

Пытаюсь быть СУХИМ
Я понимаю доходность и зависимости на поверхностном уровне, но я не уверен, что это правильный рабочий код (хотя при использовании API ошибок не выдает).
from fastapi import APIRouter, Depends, HTTPException
from tasks.models import TaskModelIn, TaskModelOut
from tasks.tables import Task

async def transaction():
"""SQLite transaction handler

> Only required for write operations that follow read operations.

We perform a SELECT first, but as it's an IMMEDIATE transaction,
we can later perform writes without getting a database locked
error.
"""
from piccolo.engine.sqlite import TransactionType

DB = Task._meta.db

async with DB.transaction(
transaction_type=TransactionType.immediate
) as transaction:
yield transaction

@task_router.put(
"/{task_id}/put/",
response_model=TaskModelOut,
dependencies=[Depends(transaction)] # Adds the transaction dependency
)
async def update_task_put(task_id: int, data: TaskModelIn):
"""Update a task with new data (full `Task` json required)

Without a transaction here you may get errors doing writes after a `.select()`.
"""

list = await Task.select().where(Task.id == task_id) # Returns `List dict`

if len(list) == 1:
# We can discard the `list` after checking it's a singleton
update = (
await Task.update(**data.model_dump())
.where(Task.id == task_id)
.returning(*Task.all_columns())
)

return update[0]

raise HTTPException(
status_code=404,
detail=f"""
Task with ID: {task_id} does not exist, too many entries found,
or the transaction went badly.
"""
)

Исходный код, отличный от DRY
Цель состоит в том, чтобы превратить это с помощью кода в функцию многократного использования, которую я могу вызвать с помощью Depends() для кода SQL любой функции (когда требуется НЕМЕДЛЕННО) без необходимости повторяться.
from piccolo.engine.sqlite import TransactionType

@task_router.put("/{task_id}/put/", response_model=TaskModelOut)
async def update_task_put(task_id: int, data: TaskModelIn):
async with Band._meta.db.transaction(
transaction_type=TransactionType.immediate
):
# With this coding style, the `with` block must be in every function
# we need to perform this task (non-DRY).

# Same code as the above "Trying to be DRY" sample ...
list = ...

if len(list) == 1:
...



Подробнее здесь: https://stackoverflow.com/questions/798 ... pends-work