如何在FastAPI请求方法外对get_db实现依赖注入？

要在CLI这类非请求处理场景中复用get_db()的依赖逻辑，同时方便测试时替换数据库会话，你可以采用以下几种方案：

方案1：封装生成器为会话获取函数 #

既然get_db()是异步生成器，我们可以封装一个工具函数来直接获取它生成的会话，避免在业务代码中写async for：

async def get_db_session() -> AsyncSession:
    async for session in get_db():
        return session

然后在CLI命令中调用这个函数：

@cli.command(name="create_superuser")
async def create_superuser():
    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

    # 直接获取会话
    db = await get_db_session()
    user = schemas.UserAdminCreate(
        username=username,
        email=None if not email else email,
        password=password,
        role="admin",
    )
    await crud.create_user(db=db, user=user)

测试适配：测试时只需mockget_db_session()，让它返回测试环境的数据库会话即可，无需修改业务代码。

方案2：使用FastAPI内置的依赖解析工具 #

FastAPI提供了resolve_dependency工具函数，可以在无请求上下文的场景下解析依赖：

from fastapi.dependencies.utils import resolve_dependency
from fastapi import Depends

@cli.command(name="create_superuser")
async def create_superuser():
    # 输入逻辑不变
    if password != confirm_password:
        click.echo("Passwords do not match")
        return

    # 解析get_db依赖
    db = await resolve_dependency(Depends(get_db), request=None, body=None)
    user = schemas.UserAdminCreate(
        username=username,
        email=None if not email else email,
        password=password,
        role="admin",
    )
    await crud.create_user(db=db, user=user)

这里request和body传None即可，因为CLI场景不需要请求上下文。

测试适配：可以通过替换get_db()的实现，或者在测试时注入自定义的依赖解析逻辑，来使用测试数据库会话。

方案3：引入依赖注入容器（适合复杂场景） #

如果你的项目有较多跨场景的依赖注入需求，可以使用第三方容器库（如injector）来统一管理依赖：

1. 先安装依赖：

pip install injector

2. 定义依赖模块：

from injector import Injector, Module, provider, singleton

class DatabaseModule(Module):
    @provider
    @singleton
    async def provide_async_session(self) -> AsyncSession:
        async for session in get_db():
            return session

# 初始化注入器
injector = Injector([DatabaseModule()])

3. 在CLI中使用注入器获取会话：

@cli.command(name="create_superuser")
async def create_superuser():
    # 输入逻辑不变
    if password != confirm_password:
        click.echo("Passwords do not match")
        return

    # 从注入器获取会话
    db = await injector.get(AsyncSession)
    user = schemas.UserAdminCreate(
        username=username,
        email=None if not email else email,
        password=password,
        role="admin",
    )
    await crud.create_user(db=db, user=user)

测试适配：测试时可以创建一个测试用的DatabaseModule，返回测试数据库会话，替换注入器中的模块即可实现依赖替换。

内容的提问来源于stack exchange，提问作者Benyamin Jafari