Wie funktionieren Transaktionen mit FastAPI Depends()? ⇐ Python

[Anonymous]

Mit Piccolo ORM verwende ich FastAPI Depends(), um Wiederholungen zu reduzieren. Die FastAPI-Seite, auf der ich diesen Code basiert, befindet sich irgendwo in der Dokumentation. SQLite erfordert für bestimmte SQL-Anweisungskombinationen einen anderen Transaktionstyp und ich möchte die with-Anweisung lieber nicht in jeder einzelnen Funktion haben.
Was machen Depends() und yield eigentlich hier? Wird die IMMEDIATE-Transaktion ordnungsgemäß erstellt? Wenn ich den Code falsch verstanden habe, gibt es eine andere Möglichkeit, Wiederholungen zu reduzieren?
Ich versuche, trocken zu sein
Ich verstehe Ertrag und Abhängigkeiten, bin mir aber nicht sicher, ob es sich um korrekten Arbeitscode handelt (obwohl er bei Verwendung der API keine Fehler verursacht):

Code: Select all

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.
"""
)

Originaler, nicht trockener Code
Das Ziel besteht darin, diesen mit-Code in eine wiederverwendbare Funktion umzuwandeln, die ich mit Depends() für den SQL-Code jeder Funktion aufrufen kann (wenn IMMEDIATE erforderlich ist), ohne mich wiederholen zu müssen:

Code: Select all

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:
...