如何在FastAPI的非路由普通函数中使用依赖注入功能

FastAPI的Depends注解默认仅会在路由函数、路由依赖、全局依赖的场景下被自动解析，普通函数（比如cron触发的工具函数）直接写Depends不会生效，你可以根据自己的场景选择以下两种方案：

方案1：手动处理依赖生成器（推荐绝大多数场景使用） #

你现有的get_db本身是生成器函数，FastAPI的Depends(get_db)本质就是自动完成「迭代生成器拿会话、执行逻辑、触发生成器的finally关闭会话」的流程，你可以手动实现这个逻辑，完全不需要修改现有代码：

def read_user(user_id: int):
    # 初始化生成器
    db_gen = get_db()
    # 拿到数据库会话，对应get_db的yield返回值
    db = next(db_gen)
    try:
        # 执行业务逻辑
        db_user = crud.get_user(db, user_id=user_id)
        return db_user
    finally:
        # 触发get_db的finally逻辑，自动关闭会话
        try:
            next(db_gen)
        except StopIteration:
            pass

如果要简化调用，也可以把get_db包装成上下文管理器，写法更简洁：

from contextlib import contextmanager

@contextmanager
def get_db_session():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 业务函数写法
def read_user(user_id: int):
    with get_db_session() as db:
        db_user = crud.get_user(db, user_id=user_id)
        return db_user

方案2：复用FastAPI内置依赖解析器（适合复杂依赖场景） #

如果你的函数有多层子依赖、或者需要复用大量已经写好的依赖逻辑，手动处理成本太高，可以直接调用FastAPI内置的依赖解析工具，实现和路由完全一致的依赖注入效果：

from fastapi.dependencies.utils import solve_dependencies
from your_main_module import app # 导入你项目中初始化的FastAPI实例

# 原函数不需要修改，保留Depends注解即可
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    return db_user

# cron调用处的写法
async def run_cron_task(user_id: int):
    # 解析依赖
    dep_result = await solve_dependencies(
        request=None,
        dependant=read_user,
        dependency_overrides_provider=app
    )
    values, errors, *_ = dep_result
    if errors:
        raise ValueError(f"依赖解析失败：{errors}")
    # 传入解析好的依赖参数调用函数
    return read_user(user_id=user_id, **values)

如果你的cron任务是同步的，可以用asyncio.run(run_cron_task(123))来调用上述异步方法。

注意事项 #

• 不要直接在普通函数调用时依赖Depends的自动注入，Python原生函数调用不会处理这个注解的默认值，只有FastAPI的路由扫描逻辑会解析该注解
• 方案1性能更高、代码更易维护，仅依赖SQLAlchemy本身的逻辑，没有额外的框架依赖，适合90%以上的使用场景
• 不管用哪种方案，都要确保数据库会话在使用完成后被关闭，避免出现连接泄漏

内容的提问来源于stack exchange，提问作者Mudasir Zahoor