How to use Depends or similar thing as dependency injection outside of FastAPI request methods?

Question

Can anybody tell me how to use dependency injection for my get_db() outside of the FastAPI routers methods? Apparently, Depends() only covers DI in request functions.

Here's the get_db() async generator:

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

In the FastAPI router, I can simply use Depends() like this:

@router.get("/interactions", response_model=List[schemas.Interaction])
async def get_all_interactions(db: Annotated[AsyncSession, Depends(get_db)]) -> List[schemas.Interaction]:
    interactions = await crud.get_interactions(db=db)

    return [
        schemas.Interaction.model_validate(interaction) for interaction in interactions
    ]

Now, outside of the request, how can I inject the get_db within a new method and get rid of async for inside of method?

@cli.command(name="create_superuser")
async def create_superuser():  # Note: how to pass db session here as param?
    username = click.prompt("Username", type=str)
    email = click.prompt("Email (optional)", type=str, default="")
    password = getpass("Password: ")
    confirm_password = getpass("Confirm Password: ")

    if password != confirm_password:
        click.echo("Passwords do not match")
        return

    async for db in database.get_db():  # Note: remove it from here
        user = schemas.UserAdminCreate(
            username=username,
            email=None if not email else email,
            password=password,
            role="admin",
        )
        await crud.create_user(db=db, user=user)

---

PS: The reason for this requirement, is that, I'm going to write a test case for the create_superuser() function, which has its own database and respective session so it would be beneficial to me to inject the session db within any methods.

Answer 1

Eventually, I could implement a simple dependency injector to resolve the async generator and reach the db async session subsequently almost acting similarly to the Depends() in FastAPI:

import asyncclick as click
import inspect

from getpass import getpass
from . import crud, schemas, database

class Provide:
    def __init__(self, value):
        self.value = value

def inject_db(f):
    sig = inspect.signature(f)
    async def wrapper(*args, **kwargs):
        for param in sig.parameters.values():
            if isinstance(param.default, Provide):
                async for db in param.default.value():
                    kwargs[param.name] = db
                    await f(*args, **kwargs)
    return wrapper

@cli.command(name="create_superuser")
@inject_db
async def create_superuser(db: database.AsyncSession = Provide(database.get_db)):
    username = click.prompt("Username", type=str)
    email = click.prompt("Email (optional)", type=str, default="")
    password = getpass("Password: ")
    confirm_password = getpass("Confirm Password: ")

    user = schemas.UserAdminCreate(
        username=username,
        email=None if not email else email,
        password=password,
        role="admin",
    )
    await crud.create_user(db=db, user=user)