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

[Anonymous]

В Piccolo ORM я использую FastAPI Depends(), чтобы уменьшить повторение. Страница 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