FastAPI如何实现带次级依赖的动态依赖注入及返回值收集
问题原因
你当前代码的核心问题是:FastAPI的依赖注入系统只会解析参数签名里标记的Depends对象,手动调用带有Depends参数的函数时,参数不会被自动注入,拿到的只是Depends实例本身,不是依赖执行后的结果,这就是你接口返回[{"dependency":{},"use_cache":true},...]的原因。
可行解决方案
我们可以通过动态生成组合依赖函数的方式实现需求,让FastAPI主动递归解析所有选中的依赖及其次级依赖,完整可运行修改代码如下:
from random import randint from typing import List, Callable import uvicorn from fastapi import FastAPI, Depends app = FastAPI() def secondary_dep(): return randint(0, 100) def primary_dep1(random_int: int = Depends(secondary_dep)): return random_int def primary_dep2(random_int: int = Depends(secondary_dep)): return random_int # 全局依赖注册映射表,服务启动时可以通过插件入口动态新增依赖 defined_deps = { "A": primary_dep1, "B": primary_dep2 } def create_combined_dep(dep_keys: List[str]) -> Callable: selected_deps = [defined_deps[key] for key in dep_keys] # 组合依赖函数,接收所有注入完成的依赖结果 def combined_dep(*results): return list(results) # 动态修改函数签名,给每个选中的依赖添加Depends标记 from inspect import Parameter, signature original_sig = signature(combined_dep) new_params = [] for idx, dep_func in enumerate(selected_deps): new_params.append(Parameter( name=f"dep_{idx}", kind=Parameter.POSITIONAL_OR_KEYWORD, default=Depends(dep_func) )) combined_dep.__signature__ = original_sig.replace(parameters=new_params) return combined_dep # 为当前路径生成A+B的组合依赖 combined_dep_ab = create_combined_dep(["A", "B"]) @app.get("/") def root(generated_dep=Depends(combined_dep_ab)): return generated_dep if __name__ == "__main__": uvicorn.run(app)
运行上述代码后调用接口,即可得到类似[23, 76]的预期返回结果,支持任意层级的次级依赖解析。
适配认证场景的优化
针对你提到的「多个认证策略只要一个通过即可访问」的业务场景,可以直接在组合依赖内完成校验逻辑,无需把所有结果返回给路径操作:
def combined_dep(*results): from fastapi import HTTPException, status for auth_result in results: # 假设认证策略返回None代表认证失败,返回用户对象代表成功 if auth_result is not None: return auth_result raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="All authentication methods failed" )
修改后路径操作拿到的直接是通过校验的用户信息,无需额外处理。
内容的提问来源于stack exchange,提问作者Ole Schwiegert
相关产品推荐
相关产品推荐

