 与 Security() 的参数、执行时机与源码实现剖析)
FastAPI 依赖注入参考Depends() 与 Security() 的参数、执行时机与源码实现剖析【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本文基于 FastAPI 官方参考文档 docs/en/docs/reference/dependencies.md完整讲解Depends()与Security()两个核心函数的全部参数dependency、use_cache、scope、scopes、用法示例与默认值并沿着仓库源码梳理它们从参数解析、依赖树构建到请求时求解的完整调用链帮助你在生产项目中正确使用依赖注入、控制缓存与执行范围并理解 OAuth2 scopes 如何写入 OpenAPI 与自动 API 文档。1. 概览依赖是如何声明的FastAPI 的依赖注入体系主要由一个特殊函数Depends()承载它接收一个可调用对象通常是函数也可以是类实例from fastapi import Depends与Path()、Query()等参数装饰函数一样Depends()定义在 fastapi/param_functions.py 中第 2283 行起其作用是包装一个“可依赖”的 callable 并返回一个内部参数对象供框架在解析端点签名时使用# fastapi/param_functions.py def Depends( dependency: Annotated[Callable[..., Any] | None, Doc(...)] None, *, use_cache: Annotated[bool, Doc(...)] True, scope: Annotated[Literal[function, request] | None, Doc(...)] None, ) - Any: return params.Depends(dependencydependency, use_cacheuse_cache, scopescope)返回的params.Depends是一个冻结 dataclass定义在 fastapi/params.py第 746-749 行# fastapi/params.py dataclass(frozenTrue) class Depends: dependency: Callable[..., Any] | None None use_cache: bool True scope: Literal[function, request] | None None dataclass(frozenTrue) class Security(Depends): scopes: Sequence[str] | None NoneSecurity直接继承Depends只多了一个scopes字段。这个类层次关系解释了二者在行为上的唯一差异——Security额外把 OAuth2 scopes 传递给 OpenAPI。在多数场景下认证、授权都可以直接用Depends()的依赖函数实现。但当需要同时声明 OAuth2 scopes 并让这些 scopes 出现在 OpenAPI以及/docs自动 UI中时应使用Security()替代Depends()from fastapi import Security2.Depends()参数参考参数类型默认值说明dependencyCallable[..., Any] \| NoneNone一个“可依赖”的 callable如函数。不要直接调用它FastAPI 会替你调用只需把对象直接传入use_cacheboolTrue请求内首次调用后若该依赖在后续再次被声明如多个子依赖共用同一依赖结果会被复用到请求结束设为False可禁用确保同一请求内重复声明时再次执行scopeLiteral[function, request] \| NoneNone主要用于含yield的依赖控制yield前/后代码的开始与结束时机见下文 执行范围2.1 基本用法依赖声明有两种等价写法Annotated推荐或旧式默认值写法。from typing import Annotated from fastapi import Depends, FastAPI app FastAPI() async def common_parameters(q: str | None None, skip: int 0, limit: int 100): return {q: q, skip: skip, limit: limit} app.get(/items/) async def read_items(commons: Annotated[dict, Depends(common_parameters)]): return commons上例即Depends()函数 docstring 中的官方示例见 fastapi/param_functions.py。注意两点不要把依赖当函数调用。Depends(common_parameters)传入的是对象本身不是common_parameters()由 FastAPI 在请求处理时负责调用。Annotated注解与“默认值”写法二选一不能同时使用源码中对此有明确断言见下文参数解析。2.2 依赖自身的参数也是声明依赖函数的签名同样遵循 FastAPI 参数体系Path、Query、Header、Cookie、Body、Form均可用于依赖参数且依赖可以嵌套依赖子依赖从而把查询逻辑、数据库会话、用户校验等横切关注点从路径操作中剥离。3. 依赖的yield与执行范围scope对于含yield的依赖例如打开数据库会话、计时、锁定资源scope参数决定“代码在yield后何时结束执行”function依赖在路径操作函数执行前开始yield前代码在路径操作函数结束后、但在响应发回客户端之前结束yield后代码。即依赖包裹的是路径操作函数。request依赖在路径操作函数前开始与function相同但结束于响应发回客户端之后。即依赖包裹的是整个请求与响应周期含响应发送与后台任务收尾。不设置None从源码看fastapi/dependencies/models.py 的_get_computed_scope生成器依赖默认计算为request普通依赖为None无生命周期语义。from typing import Annotated from fastapi import Depends, FastAPI app FastAPI() async def verify_safeword( scope: Annotated[str | None, Depends(None, scoperequest)], ): # 此处省略实际校验逻辑 pass说明scope是Depends()的专属参数Security()的公开签名中没有scope参数见 fastapi/param_functions.py 中Security的定义仅有dependency、scopes、use_cache。一个值得注意的源码级约束在 fastapi/dependencies/utils.py 的get_dependant()中如果一个含yield的依赖自身计算作用域为request而它又声明了一个scopefunction的子依赖框架会抛出DependencyScopeError提示“作用域为 request 的依赖不能依赖作用域为 function 的依赖”。这是因为function作用域的收尾代码会在响应发送前执行无法被外层request作用域正确包裹。scope在运行时如何生效在 fastapi/dependencies/utils.py 的solve_dependencies()中use_astack request_astack if sub_dependant.scope function: use_astack function_astack solved await _solve_generator( dependantuse_sub_dependant, stackuse_astack, sub_valuessolved_result.values, )依赖生成器被包进AsyncExitStack上下文管理器_solve_generator第 566-574 行。请求作用域与函数作用域使用两个不同的栈分别存放在request.scope的fastapi_inner_astack与fastapi_function_astack中见第 601-608 行因此scopefunction的依赖其yield后代码会在函数栈关闭时路径操作函数返回后、响应发出前执行而默认/request作用域的依赖则存活到请求栈关闭响应发回客户端之后。4.use_cache请求内缓存与缓存键默认use_cacheTrue。在solve_dependencies()中的对应逻辑fastapi/dependencies/utils.pysub_dependant_cache_key _get_cache_key( dependantsub_dependant, uses_scopes_cache_uses_scopes_cache, ) if sub_dependant.use_cache and sub_dependant_cache_key in dependency_cache: solved dependency_cache[sub_dependant_cache_key] # ... 否则执行依赖并把结果写入 dependency_cache缓存键的构造在 fastapi/dependencies/models.pydef _get_cache_key( *, dependant: Dependant, uses_scopes_cache: _UsesScopesCache | None None, ) - DependencyCacheKey: scopes_for_cache ( tuple(sorted(set(_get_oauth_scopes(dependantdependant)))) if _uses_scopes(dependantdependant, cacheuses_scopes_cache) else () ) return ( dependant.call, scopes_for_cache, _get_computed_scope(dependantdependant) or , )从源码结构看缓存键是三元组(依赖 callable, scopes 排序去重后的元组, 计算作用域)。这意味着同一请求内同一依赖同一 callable、同一 scopes、同一 scope第二次出现时直接复用首次结果——这是“同一依赖在多个子依赖中重复声明只执行一次”的底层原因若两个地方对同一依赖声明了不同的 scopes例如Security(get_current_user, scopes[items])与Security(get_current_user, scopes[users:write])缓存键不同依赖会被分别执行从而为每个 scope 集合得到独立的解析结果将use_cacheFalse时sub_dependant.use_cache为假跳过命中检查保证每次声明都重新执行。Dependant模型fastapi/dependencies/models.py保存了own_oauth_scopes、parent_oauth_scopes、use_cache、scope等字段是理解上述机制的关键数据结构dataclass(slotsTrue) class Dependant: ... use_cache: bool True path: str | None None scope: Literal[function, request] | None None own_oauth_scopes: list[str] | None None parent_oauth_scopes: list[str] | None None5.Security()参数参考参数类型默认值说明dependencyCallable[..., Any] \| NoneNone同Depends()的dependency可依赖的 callable由 FastAPI 调用scopesSequence[str] \| NoneNone使用该 Security 依赖的路径操作所需的 OAuth2 scopes。“scope”一词来自 OAuth2 规范通常指权限permission或角色role这些 scopes 会集成进 OpenAPI因而在/docs等自动文档中可见use_cacheboolTrue同Depends()的use_cacheSecurity()与Depends()的唯一区别是它可以声明 OAuth2 scopes并将这些 scopes 写入 OpenAPI 与自动 UI 文档见 fastapi/param_functions.py 中Security的 docstring 示例from typing import Annotated from fastapi import FastAPI, Security app FastAPI() app.get(/users/me/items/) async def read_own_items( current_user: Annotated[User, Security(get_current_active_user, scopes[items])] ): return [{item_id: Foo, owner: current_user.username}]5.1 scopes 在请求时的传递scopes 不只是文档装饰。在 fastapi/dependencies/utils.py 中get_parameterless_sub_dependant()处理无参依赖时own_oauth_scopes: list[str] [] if isinstance(depends, params.Security) and depends.scopes: own_oauth_scopes.extend(depends.scopes) return get_dependant( pathpath, calldepends.dependency, scopedepends.scope, own_oauth_scopesown_oauth_scopes, )而在get_dependant()中每个带Security注解的参数都会把 scopes 记录到子Dependant并把当前层累积的 scopes 作为parent_oauth_scopes传给更深层第 302-331 行。合并规则见 fastapi/dependencies/models.py 的_get_oauth_scopes()——保留顺序、去重地拼接“父层 scopes 自身 scopes”。最终若依赖或路径操作的参数类型标注为SecurityScopessolve_dependencies()会把合并后的 scopes 注入fastapi/dependencies/utils.pyif dependant.security_scopes_param_name: values[dependant.security_scopes_param_name] SecurityScopes( scopes_get_oauth_scopes(dependantdependant) )于是你可以在依赖内这样读取当前累积的 scopesfrom fastapi.security import SecurityScopes async def get_current_user(security_scopes: SecurityScopes): # security_scopes.scopes: list[str]如 [items] ...5.2 scopes 如何进入 OpenAPIOpenAPI 生成时fastapi/openapi/utils.py会通过依赖树的own_oauth_scopes/parent_oauth_scopes汇总出每个操作的security定义_get_openapi_security_definitions第 132 行起使Security(..., scopes[...])声明的 scopes 出现在生成的 OpenAPI JSON 中/docsSwagger UI等界面因此能展示每个操作要求的 scope 列表并在“Authorize”按钮的 scope 勾选中体现。6. 源码剖析一条Depends()声明的完整生命周期结合仓库源码Depends()/Security()的处理可以分成四步。6.1 参数解析analyze_paramfastapi/dependencies/utils.py 的analyze_param()负责从端点/依赖函数签名中识别Dependsfastapi_annotations [ arg for arg in annotated_args[1:] if isinstance(arg, (FieldInfo, params.Depends)) ] ... # Get Annotated Depends elif isinstance(fastapi_annotation, params.Depends): depends fastapi_annotation # Get Depends from default value if isinstance(value, params.Depends): assert depends is None, ( Cannot specify Depends in Annotated and default value f together for {param_name!r} ) ... depends value两条要点Depends支持两种位置Annotated注解内推荐或参数默认值旧式两者不可同时出现源码会直接断言失败若Depends()未显式传入 callabledepends.dependency is None源码会用类型注解作为依赖对象第 467-471 行depends dataclasses.replace(depends, dependencytype_annotation)。这支持Annotated[User, Depends(use_cacheFalse)]这类写法。6.2 构建依赖树get_dependantget_dependant()fastapi/dependencies/utils.py递归地把每个带Depends的参数转换为子Dependant形成依赖树提取Annotated中的Depends/Security若为Security则把scopes记入own_oauth_scopes把use_cache、scope透传给子Dependant若依赖参数类型是Request、Response、BackgroundTasks、SecurityScopes等特殊类型则记为对应的注入名add_non_field_param_to_dependency第 350-371 行由框架直接注入。6.3 请求时求解solve_dependencies请求到来时solve_dependencies()递归求解依赖树第 586-731 行关键行为依赖覆盖override若dependency_overrides_provider.dependency_overrides中存在原 callable 的替换第 623-638 行则用替换 callable 重新构建Dependant并求解——这是app.dependency_overrides测试机制的底层实现缓存命中按use_cache与缓存键决定是否复用见 第 4 节执行方式生成器依赖走_solve_generator并进入对应作用域的AsyncExitStack协程依赖await call(...)同步依赖通过run_in_threadpool在线程池中执行第 673-676 行保证不阻塞事件循环特殊注入Request/WebSocket/HTTPConnection/Response/BackgroundTasks/SecurityScopes等按参数名注入第 709-724 行结果通过values[sub_dependant.name] solved传回父层最终作为路径操作函数的入参。6.4Scope与生成器依赖的时序含yield的依赖通过contextlib.contextmanager/asynccontextmanager包装fastapi/dependencies/utils.py进入AsyncExitStack。scope参数或生成器默认的request决定它进入哪个栈function→function_astack在路径操作函数返回后、响应发送前关闭yield后代码先执行request生成器依赖默认→request_astack在整个请求-响应周期结束后关闭yield后代码晚执行。这与scope参数的官方文档描述完全一致function包裹“路径操作函数”request包裹“请求与响应周期”。7. 常见错误与行为要点基于源码中的断言与校验逻辑整理以下实践要点场景行为源码依据Annotated与默认值中同时写Depends触发断言错误二者只能取其一analyze_paramrequest作用域生成器依赖声明了function作用域子依赖抛出DependencyScopeErrorget_dependant同一依赖在不同scopes下被多次声明缓存键不同依赖分别执行_get_cache_keyuse_cacheFalse跳过缓存命中检查每次声明都执行solve_dependencies同步非 async依赖在线程池中执行不阻塞事件循环solve_dependencies此外Depends()的 callable 不限于函数从analyze_param对dataclasses.replace(depends, dependencytype_annotation)的处理以及get_parameterless_sub_dependant对类调用的支持来看依赖也可以是实现了__call__的类实例便于携带状态或依赖注入的配置对象Security()同理。8. 小结Depends()是 FastAPI 依赖注入的核心声明函数参数为dependency、use_cacheTrue、scopeNoneSecurity()在其基础上增加scopes参数用于把 OAuth2 scopes 写入 OpenAPI 与自动文档。从源码看params.Depends/params.Security是轻量冻结 dataclassfastapi/params.py真正的工作发生在依赖树构建get_dependant与请求求解solve_dependencies两个阶段。use_cache基于(call, scopes, scope)三元组缓存键实现请求内去重执行scope基于两个AsyncExitStack实现yield依赖的差异化生命周期scopes通过own_oauth_scopes/parent_oauth_scopes自底向上合并最终注入SecurityScopes并进入 OpenAPI。相关参考文档docs/en/docs/reference/dependencies.md、docs/en/docs/tutorial/dependencies/依赖教程、docs/en/docs/advanced/security/oauth2-scopes/OAuth2 scopes 进阶。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考