I'm trying to use FastAPI Depends() as a way to reduce repetition (with Piccolo ORM). I can't find the original FastAPI page I based this code on, but it's somewhere in the documentation. SQLite requires a different transaction type for certain SQL statement combinations, and I'd rather not have the with statement in every single function.

Sample code is below, but I'd like to understand:

1. What is Depends() and yield actually doing here?
   • Is it creating the IMMEDIATE transaction properly, or is it not working as I assumed?
2. If I've got the code wrong, is there any other way to reduce repetition?

Trying to be DRY

I understand yield and dependencies on a superficial level, but I'm not sure this is correct working code (although it gives no errors when using the 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.
        """
    )

Original non-DRY code

The aim is to covert this with code into a reusable function I can call with Depends() for any function's SQL code (when requiring IMMEDIATE) without having to repeat myself.

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