1. 作用
路径函数很少能够独立完成工作。一个聊天接口可能需要读取配置、确认当前用户、取得数据库 Session,再创建业务服务。如果每个接口都手动完成这些步骤,代码很快就会变成这样:
01@app.post("/chat")02async def chat(payload: ChatRequest):03user = get_current_user()04session = create_session()05service = ChatService(model, session)0607try:08return await service.answer(user, payload.question)09finally:10await session.close()
这段代码不是不能运行,问题在于每个路由都要知道对象如何创建、以什么顺序调用,以及何时清理。身份校验、数据库管理和服务组装会不断重复,测试时也很难只替换其中一个环节。
FastAPI 的依赖注入就是为了解决这个问题。我们先用一句话理解它:
路径函数只声明自己需要什么,FastAPI 负责先取得这些对象,再把它们传给路径函数。
本文的完整可请求案例位于 src/fastapi/api/deps/,每个下划线命名的 Python 文件对应一个独立路由模块;它们统一由 deps/__init__.py 聚合。文章中的调用片段、依赖别名片段、外部数据库或客户端示例,以及测试代码不对应独立接口文件。
deps/__init__.py 负责声明依赖案例的领域前缀并聚合 7 个完整案例,上层 api/router.py 再将整个依赖模块挂载到 /api/v1:
01from fastapi import APIRouter0203from . import (04callable_dependency,05class_dependency,06dependency_cache,07first_dependency,08pagination,09path_dependency,10sub_dependencies,11)1213# /api/v1 由上层聚合路由统一添加,本模块只声明自己的领域前缀。14router = APIRouter(prefix="/deps", tags=["dependencies"])1516router.include_router(first_dependency.router)17router.include_router(pagination.router)18router.include_router(path_dependency.router)19router.include_router(sub_dependencies.router)20router.include_router(dependency_cache.router)21router.include_router(class_dependency.router)22router.include_router(callable_dependency.router)
因此,完整案例中的相对路径会统一组合成 /api/v1/deps/...。
在 FastAPI 中,依赖通常就是一个普通的 Python 函数。下面的路径函数需要一个模型名称,因此把 get_model_name 声明为依赖:
先发送一次模型信息请求,观察依赖函数的返回值如何进入路径函数:
01from typing import Annotated0203from fastapi import APIRouter, Depends0405router = APIRouter()060708def get_model_name() -> str:09return "deepseek-v4-flash"101112@router.get("/model")13async def model_info(14model_name: Annotated[str, Depends(get_model_name)],15) -> dict[str, str]:16return {"model": model_name}
浏览器访问 /model 时,实际执行过程如下:
- FastAPI 发现
model_name依赖get_model_name; - FastAPI 调用
get_model_name(); - 函数返回
"deepseek-v4-flash"; - FastAPI 把返回值传给
model_name; - 所有参数准备完成后,FastAPI 才调用
model_info()。
因此,我们没有在路径函数中亲自调用 get_model_name(),但 model_name 仍然得到了它的返回值。这里的依赖是 get_model_name,注入则是 FastAPI 把返回值交给路径函数的过程。
依赖函数不会在模块导入时自动执行,而是在处理需要它的请求时执行。它也不是一种特殊函数:离开 FastAPI 后,我们仍然可以像普通函数一样直接调用和测试它。
2. 语法
第一次看到下面这段代码时,最容易困惑的是 Annotated 与 Depends 各自负责什么:
1model_name: Annotated[str, Depends(get_model_name)]
可以把它拆成四部分理解:
| 部分 | 含义 |
|---|---|
model_name | 路径函数最终使用的参数名 |
str | model_name 的 Python 类型 |
Depends(...) | 告诉 FastAPI 这个值来自依赖 |
get_model_name | FastAPI 需要调用的依赖函数 |
Annotated 是 Python 标准库提供的类型工具,可以在一个类型后面附加额外元数据。对编辑器和类型检查器来说,model_name 仍然是 str;对 FastAPI 来说,Depends(get_model_name) 又提供了“这个字符串从哪里来”的信息。
传给 Depends 的是函数本身,所以不能加括号:
1# 正确:把函数交给 FastAPI,等待请求到来后再调用2Depends(get_model_name)34# 错误:模块加载时就执行函数,并把返回的字符串传给 Depends5Depends(get_model_name())
旧写法把 Depends 放在参数默认值的位置:
1async def model_info(2model_name: str = Depends(get_model_name),3):4...
它目前仍然可以使用,但最新官方示例更推荐 Annotated。这种写法把类型和 FastAPI 元数据分开得更清楚,也方便提取成可复用别名:
1from typing import Annotated23from fastapi import Depends45ModelNameDep = Annotated[str, Depends(get_model_name)]67@app.get("/model")8async def model_info(model_name: ModelNameDep):9return {"model": model_name}
ModelNameDep 只是普通的 Python 类型别名,不会保存某次请求的模型名称。每次请求到来时,FastAPI 仍然会重新解析其中的 Depends(get_model_name)。
3. 参数
依赖函数的参数解析方式和路径函数基本相同。它可以读取查询参数、路径参数、请求头、Cookie、请求体,也可以使用 Pydantic 完成校验。
例如,多个列表接口都需要相同的分页规则,我们可以把它提取成依赖:
修改 offset 和 limit 后发送请求,可以直接观察查询参数经过依赖解析后的结果:
01from typing import Annotated0203from fastapi import APIRouter, Depends, Query04from pydantic import BaseModel0506router = APIRouter()070809class Pagination(BaseModel):10offset: int11limit: int121314def get_pagination(15offset: Annotated[int, Query(ge=0)] = 0,16limit: Annotated[int, Query(ge=1, le=100)] = 20,17) -> Pagination:18return Pagination(offset=offset, limit=limit)192021PaginationDep = Annotated[Pagination, Depends(get_pagination)]222324@router.get("/conversations")25async def list_conversations(paging: PaginationDep) -> Pagination:26return paging
访问 /conversations?offset=20&limit=10 时,FastAPI 会先从查询字符串中取得两个值,完成整数转换和范围校验,再调用:
1get_pagination(offset=20, limit=10)
路径函数最终收到的是 Pagination(offset=20, limit=10),而不是两个未经处理的字符串。如果 limit=0,校验会在路径函数执行前失败,并返回 422。
依赖中的 Query、Header 和 Pydantic 规则也会合并到当前路由的 OpenAPI 文档。因此,提取依赖不会让 /docs 丢失参数说明。
依赖还可以直接取得路径参数。只要依赖参数名与路由中的路径参数一致,FastAPI 就会把同一个值交给它:
下面默认查询存在的 conversation-1。也可以修改 ID,观察依赖抛出的 404 如何阻止路径函数继续执行:
01from typing import Annotated0203from fastapi import APIRouter, Depends, HTTPException, status0405router = APIRouter()0607conversations = {08"conversation-1": {"title": "FastAPI"},09}101112def get_conversation(13conversation_id: str,14) -> dict[str, str]:15conversation = conversations.get(conversation_id)1617if conversation is None:18raise HTTPException(19status_code=status.HTTP_404_NOT_FOUND,20detail="会话不存在",21)2223return conversation242526ConversationDep = Annotated[27dict[str, str],28Depends(get_conversation),29]303132@router.get("/conversations/{conversation_id}")33async def read_conversation(34conversation: ConversationDep,35) -> dict[str, str]:36return conversation
如果路由路径是 /conversations/{conversation_id},这个依赖就能使用其中的 conversation_id。依赖并不是从路径函数内部读取变量,而是和路径函数一样,由 FastAPI 从当前请求中解析参数。
4. 子依赖
一个依赖还可以继续依赖另一个依赖,这叫作子依赖。认证流程很适合用来理解这层关系:先从请求头取得 API Key,再根据 API Key 找到用户,最后才执行接口。
使用默认 API Key 可以取得用户;修改为错误值后,响应会变成 401:
01from typing import Annotated0203from fastapi import APIRouter, Depends, Header, HTTPException, status04from pydantic import BaseModel0506router = APIRouter()070809class User(BaseModel):10id: str11name: str121314def get_api_key(15x_api_key: Annotated[16str | None,17Header(alias="X-API-Key"),18] = None,19) -> str:20if x_api_key != "local-demo-key":21raise HTTPException(22status_code=status.HTTP_401_UNAUTHORIZED,23detail="API Key 无效",24headers={"WWW-Authenticate": "APIKey"},25)2627return x_api_key282930ApiKeyDep = Annotated[str, Depends(get_api_key)]313233def get_current_user(api_key: ApiKeyDep) -> User:34return User(id="user-1", name="小微")353637CurrentUserDep = Annotated[User, Depends(get_current_user)]383940@router.get("/me")41async def read_me(user: CurrentUserDep) -> User:42return user
read_me 只声明了 get_current_user,但 FastAPI 会递归发现它还需要 get_api_key,于是形成下面的调用链:
1请求头2-> get_api_key3-> get_current_user4-> read_me
如果 API Key 缺失或错误,get_api_key 会抛出异常,后面的 get_current_user 和 read_me 都不会执行。依赖校验通过后,API Key 只在依赖链内部使用,接口最终返回的用户对象不会意外包含密钥。
FastAPI 会根据这些声明建立依赖图。不要依赖几个无关参数在函数签名中的书写顺序来安排业务流程;如果 B 必须在 A 之后执行,就应该让 B 显式依赖 A。
5. 缓存
同一个请求中,一个依赖可能通过不同路径被使用多次。FastAPI 默认会执行一次并复用返回值。
发送请求时可以同时观察页面响应与本地 FastAPI 终端,确认 load_user() 虽然被两条依赖路径使用,但当前请求只打印一次:
01from typing import Annotated0203from fastapi import APIRouter, Depends0405router = APIRouter()060708def load_user() -> dict[str, str]:09print("当前请求只执行一次")10return {"id": "user-1", "name": "小微"}111213UserDep = Annotated[dict[str, str], Depends(load_user)]141516def check_plan(user: UserDep) -> str:17return "pro"181920PlanDep = Annotated[str, Depends(check_plan)]212223@router.get("/account")24async def account(25user: UserDep,26plan: PlanDep,27) -> dict[str, str | dict[str, str]]:28return {"user": user, "plan": plan}
这里有两条路径需要 load_user:
account直接需要user;check_plan也需要user。
FastAPI 在当前请求中只调用一次 load_user(),然后把同一个结果交给两处使用。这种行为可以避免重复查询数据库、重复解析 Token,或者重复创建请求级资源。
这个缓存只存在于一次请求的依赖解析过程。第二个请求到来时,load_user() 仍然会再次执行,它不是 Redis,也不是跨请求内存缓存。
少数场景确实需要每次都重新计算,可以在使用依赖的位置关闭缓存:
1FreshUserDep = Annotated[2dict[str, str],3Depends(load_user, use_cache=False),4]
不要仅仅为了“确保数据最新”就给数据库 Session、当前用户等依赖加上 use_cache=False。同一请求重复创建资源通常会增加消耗,还可能让不同依赖拿到彼此不一致的对象。因为默认缓存会复用同一个对象,可变对象也不应该被某个依赖随意修改。
6. yield
普通依赖使用 return 提供一个值,但数据库 Session、文件句柄等资源还需要在使用结束后关闭。此时可以把 return 换成 yield,并把清理逻辑写在它后面。
01from collections.abc import AsyncIterator02from typing import Annotated0304from fastapi import Depends, FastAPI05from sqlalchemy.ext.asyncio import AsyncSession0607from database import async_session_factory08from models import User0910app = FastAPI()1112async def get_session() -> AsyncIterator[AsyncSession]:13async with async_session_factory() as session:14yield session1516SessionDep = Annotated[AsyncSession, Depends(get_session)]1718@app.get("/users/{user_id}")19async def get_user(user_id: int, session: SessionDep):20return await session.get(User, user_id)
yield session 可以理解为:先把 Session 暂时交给后面的代码使用,等它们结束后,再回到这里完成清理。 async with 离开时会关闭 Session,并把数据库连接归还给连接池。
按照默认的 scope="request",一次普通请求的执行过程是:
- 进入
get_session,创建 Session; - 执行到
yield,把 Session 注入路径函数; - 路径函数查询用户并返回结果;
- FastAPI 创建并发送响应;
- 回到
yield后面,退出async with并清理 Session。
即使路径函数抛出异常,清理步骤也会执行。这和 Python 上下文管理器的行为相同。如果依赖需要显式捕获异常,应使用 finally 保证清理;如果它不能真正处理异常,还必须继续 raise,不能把原异常吞掉。
数据库连接池或模型客户端通常在应用的 lifespan 中创建一次并长期复用,请求级 Session 则由 yield 依赖为每个请求创建和关闭。不要在每次请求中重新创建整个连接池。
当前 FastAPI 还支持把带 yield 的依赖设为 scope="function":
1SessionDep = Annotated[2AsyncSession,3Depends(get_session, scope="function"),4]
两种作用域的区别在于何时执行 yield 后面的清理代码:
scope | 清理时机 | 适合场景 |
|---|---|---|
"request" | 响应发送完成后,默认值 | 流式响应期间仍然需要资源 |
"function" | 路径函数返回后、响应发送前 | 响应阶段不再需要资源 |
例如,StreamingResponse 的生成器还需要不断读取数据库时,必须让 Session 活到流结束,不能使用 scope="function" 提前关闭。反过来,如果 Session 只用于路径函数中的一次身份查询,就可以考虑提前清理。
FastAPI 过去调整过 yield 依赖的退出时机,当前规则应以 scope 为准。scope="function" 是 FastAPI 0.121.0 加入的能力;使用它之前应确认项目版本。
多个 yield 依赖嵌套时,进入顺序是从子依赖到外层依赖,清理顺序则相反。默认 request scope 的依赖如果在清理阶段还可能使用子依赖,那么它的 yield 子依赖也必须存活到请求结束,FastAPI 会检查这类不合法的作用域组合。
7. 异步
依赖可以写成普通 def,也可以写成 async def,并且可以与同步或异步路径函数自由组合。FastAPI 会选择相应的执行方式:
| 写法 | FastAPI 如何执行 | 常见场景 |
|---|---|---|
def | 在线程池中调用 | 同步 SDK、同步文件或数据库操作 |
async def | 在事件循环中直接等待 | 异步数据库、异步 HTTP、异步模型调用 |
选择的关键不是“项目是不是异步项目”,而是依赖内部调用的库是否支持异步。假设一个依赖内部调用异步 HTTP 客户端,就应该写成:
1async def load_profile() -> Profile:2response = await http_client.get("/profile")3return Profile.model_validate(response.json())
如果使用同步 SDK,则可以保留普通函数,让 FastAPI 把它放到线程池:
1def load_settings() -> Settings:2return settings_loader.load()
最危险的情况是在 async def 中直接调用耗时的同步 IO。函数中虽然出现了 async,同步调用仍然会阻塞事件循环,影响同一进程里的其他请求。应该改用异步客户端,或者把无法替换的同步工作放入线程池。
CPU 密集型计算也不适合长期占用事件循环或普通请求线程。图片处理、大规模数据计算等任务通常应该交给独立进程或任务队列。依赖注入只负责组织调用关系,不会自动解决阻塞和 CPU 竞争。
8. 类
Depends() 接受的是 callable,也就是“可以被调用的对象”。函数是 callable,类也是 callable,因为可以通过 Pagination(...) 创建实例。
下面把一个类本身作为依赖:
这个请求仍然使用分页参数,但这一次负责解析参数和保存结果的是 Pagination 类实例:
01from typing import Annotated0203from fastapi import APIRouter, Depends, Query0405router = APIRouter()060708class Pagination:09def __init__(10self,11offset: Annotated[int, Query(ge=0)] = 0,12limit: Annotated[int, Query(ge=1, le=100)] = 20,13):14self.offset = offset15self.limit = limit161718PaginationDep = Annotated[Pagination, Depends()]192021@router.get("/items")22async def list_items(paging: PaginationDep) -> dict[str, int]:23return {24"offset": paging.offset,25"limit": paging.limit,26}
这里的 Depends() 没有显式参数,FastAPI 会从 Annotated 中的 Pagination 类型推断依赖目标。它检查 Pagination.__init__() 的参数,解析 offset 和 limit,然后创建并注入一个 Pagination 实例。
另一种情况是先创建一个配置好的实例,再让 FastAPI 调用它的 __call__():
修改查询文本,可以观察配置为 fastapi 的可调用实例如何返回布尔结果:
01from typing import Annotated0203from fastapi import APIRouter, Depends, Query0405router = APIRouter()060708class ContainsText:09def __init__(self, keyword: str):10self.keyword = keyword1112def __call__(13self,14q: Annotated[str, Query()] = "",15) -> bool:16return self.keyword.lower() in q.lower()171819contains_fastapi = ContainsText("fastapi")2021ContainsFastAPIDep = Annotated[22bool,23Depends(contains_fastapi),24]252627@router.get("/search")28async def search(contains_keyword: ContainsFastAPIDep) -> dict[str, bool]:29return {"contains_keyword": contains_keyword}
ContainsText("fastapi") 在应用加载时执行一次,用于保存固定配置;每次访问 /search?q=... 时,FastAPI 调用的是 contains_fastapi.__call__(),并把布尔结果传给 contains_keyword。这种写法适合可配置校验器。
模块级实例会被多个请求共享,所以不要把当前用户、临时查询结果等请求状态写进它的 self 属性。固定配置可以共享,请求数据应该通过 __call__() 参数传入并通过返回值传出。
9. 服务
依赖注入很适合把业务服务交给路径函数。下面假设 lifespan 已经把模型客户端保存到 app.state.model;ChatService 负责模型调用,路径函数只负责接收 HTTP 数据并返回结果:
01from typing import Annotated0203from fastapi import Depends, FastAPI, Request04from pydantic import BaseModel0506app = FastAPI()0708class ChatRequest(BaseModel):09question: str1011class ChatService:12def __init__(self, model):13self.model = model1415async def answer(self, question: str) -> str:16message = await self.model.ainvoke(question)17return str(message.content)1819def get_chat_service(request: Request) -> ChatService:20return ChatService(request.app.state.model)2122ChatServiceDep = Annotated[23ChatService,24Depends(get_chat_service),25]2627@app.post("/chat")28async def chat(29payload: ChatRequest,30service: ChatServiceDep,31):32answer = await service.answer(payload.question)33return {"answer": answer}
这里包含两种生命周期:
request.app.state.model通常由 lifespan 在应用启动时创建,供多个请求共享;ChatService是轻量包装,由依赖在请求中创建,并在当前请求中缓存复用。
FastAPI 不会看到 ChatService 类型后自动寻找实现。真正建立关系的是 Depends(get_chat_service),它明确告诉 FastAPI 应该如何得到服务。
把业务服务作为依赖的主要价值是解耦组装方式。生产环境可以注入真实模型,测试环境可以注入假服务,而路径函数不需要改变。服务内部的核心业务流程仍然应该通过普通方法明确调用,不要把整段聊天逻辑隐藏成一个难以发现的依赖。
10. 范围
依赖可以放在参数上,也可以注册到路径、Router 或整个应用。区别主要在于适用范围,以及是否需要使用依赖返回值。
01from fastapi import APIRouter, Depends, FastAPI0203from auth import require_admin, require_user04from request_context import check_request_id0506app = FastAPI(07dependencies=[Depends(check_request_id)],08)0910admin_router = APIRouter(11prefix="/admin",12dependencies=[Depends(require_user)],13)1415@admin_router.get(16"/reports",17dependencies=[Depends(require_admin)],18)19async def reports():20return {"items": []}2122app.include_router(admin_router)
可以按下面的方式选择:
| 声明位置 | 生效范围 | 是否直接取得返回值 |
|---|---|---|
| 函数参数 | 当前路径函数 | 可以 |
| 路径装饰器 | 当前路径函数 | 不使用返回值 |
APIRouter | Router 中的一组路由 | 不使用返回值 |
FastAPI | 应用中的所有路径操作 | 不使用返回值 |
如果路径函数需要当前用户对象,就应该把依赖声明成参数。如果只要求“检查通过后才能继续”,例如校验管理员权限,放在 dependencies=[...] 中会更清楚。即使返回值没有被接收,这些依赖仍然会执行,也仍然可以抛出异常。
应用级依赖要谨慎使用,因为它会影响所有路径操作。某项规则只属于后台接口时,更适合挂在对应的 APIRouter 上,而不是让每个公开接口都承担它。
认证、权限、请求级资源和重复参数解析适合使用依赖。请求耗时统计、统一响应头和 CORS 等需要包围整个请求响应过程的逻辑,更适合中间件。
11. 测试
依赖把“使用什么”和“如何得到它”分开以后,测试就可以替换后者。FastAPI 使用 app.dependency_overrides 保存测试覆盖关系,键是原始依赖 callable,值是替代 callable。
01from fastapi.testclient import TestClient0203from main import app04from services import get_chat_service0506class FakeChatService:07async def answer(self, question: str) -> str:08return f"测试回复:{question}"0910def get_fake_chat_service() -> FakeChatService:11return FakeChatService()1213client = TestClient(app)1415def test_chat_without_real_model():16app.dependency_overrides[get_chat_service] = (17get_fake_chat_service18)1920try:21response = client.post(22"/chat",23json={"question": "你好"},24)25finally:26app.dependency_overrides.pop(27get_chat_service,28None,29)3031assert response.status_code == 20032assert response.json() == {33"answer": "测试回复:你好",34}
覆盖生效后,FastAPI 调用 get_fake_chat_service,不会再调用原来的 get_chat_service,原依赖下面的子依赖也不会执行。这样,测试不会连接真实模型,也不会产生外部费用。
覆盖的键必须是写进 Depends() 的原始对象,而不是 ChatServiceDep 类型别名。测试结束后要恢复覆盖,避免影响同一进程中的其他测试。测试数量增多后,可以用 pytest fixture 统一完成设置和清理。
依赖函数本身仍然是普通函数,因此分页计算、权限判断等纯逻辑也可以脱离 FastAPI 单独测试。只有需要验证请求解析和整条依赖图时,才需要通过 TestClient 发起请求。
12. 边界
依赖很方便,但它不是所有初始化和横切逻辑的统一入口。可以根据生命周期和职责做出选择:
| 机制 | 适合处理 |
|---|---|
| 依赖 | 当前请求需要的参数、身份、Session 和业务服务 |
| 路径函数 | HTTP 输入输出与一次业务操作的编排 |
| 中间件 | Request ID、耗时、CORS 等请求响应横切逻辑 |
| lifespan | 模型客户端、连接池等应用级资源的启动与关闭 |
| 后台任务 | 响应发送后仍需继续执行的工作 |
工程中还要注意以下边界:
- FastAPI 不会扫描项目并自动注入任意类,只有显式声明在
Depends()中的 callable 才会进入依赖图; - 依赖应完成一个容易描述的职责,不要隐藏扣款、发消息等核心业务副作用;
- 请求级可变数据不要存入模块全局对象或共享的 callable 实例;
- 创建了资源,就要用
yield、上下文管理器或 lifespan 明确清理; - 依赖之间不能循环引用,否则 FastAPI 无法建立可执行的依赖图;
- 不要用
use_cache=False、同步函数或全局依赖掩盖不清晰的生命周期设计。
判断一段逻辑是否适合成为依赖,可以问两个问题:它是不是路径函数执行前必须准备的对象或条件?路径函数是否需要直接使用它的结果? 如果答案都是肯定的,参数依赖通常很合适;如果逻辑需要包围所有请求、持续到应用关闭或在响应后执行,就应该选择其他机制。