---
title: FastAPIdependency-injection完全指南 - 代码复用与权限校验最佳实践 | 道满PythonAI
description: 深入掌握FastAPI依赖注入核心机制，实现代码复用、权限校验与数据库连接。包含实战示例、依赖链构建及企业级应用模式。
keywords: [FastAPI, 依赖注入, DI, 代码复用, 权限校验, 数据库连接, Depends, 认证系统, 中间件]
date: 2026-04-10
updated: 2024-01-15
author: 道满PythonAI
tags: [FastAPI, 依赖注入, 认证授权, 企业级开发]
---
# FastAPIdependency-injection完全指南
> 📂 所属阶段：第二阶段 — 进阶黑科技（核心篇）
> 🔗 相关章节：[FastAPI异步编程深度解析](/pyweb/fastapi/async-await-principles) · [FastAPImiddleware-application](/pyweb/fastapi/middleware-application)
依赖注入（Dependency Injection）是 FastAPI 最吸引人的特性之一。它让你用**声明式**的方式告诉系统“我需要什么”，而不再重复编写那些散落在各处的样板代码。本文将带你从零开始，掌握依赖注入的各类用法，并构建一套可重用的认证授权体系。
## 目录
- [什么是依赖注入？](#什么是依赖注入)
- [基础用法](#基础用法)
- [数据库连接与依赖](#数据库连接与依赖)
- [依赖链与级联](#依赖链与级联)
- [可选依赖与默认值](#可选依赖与默认值)
- [依赖缓存机制](#依赖缓存机制)
- [实战：认证授权体系](#实战认证授权体系)
- [进阶用法](#进阶用法)
- [避坑指南与性能优化](#避坑指南与性能优化)
- [总结](#总结)
---
## 什么是依赖注入？
### 不用依赖注入的痛点
假设我们需要在多个路由中做用户认证，一种很自然的写法是直接把认证逻辑粘贴进去：
```python
# ❌ 没有依赖注入：重复逻辑散落在每个路由
from fastapi import FastAPI, HTTPException, Request
from db import fake_db
app = FastAPI()
@app.get("/users/me")
def get_current_user(request: Request):
token = request.headers.get("Authorization")
if not token or not token.startswith("Bearer "):
raise HTTPException(401, "Unauthorized")
user = fake_db.get_user(token.replace("Bearer ", ""))
if not user:
raise HTTPException(401, "Invalid token")
return user
@app.get("/orders")
def get_orders(request: Request):
token = request.headers.get("Authorization")
# 重复的认证逻辑！
if not token or not token.startswith("Bearer "):
raise HTTPException(401, "Unauthorized")
user = fake_db.get_user(token.replace("Bearer ", ""))
if not user:
raise HTTPException(401, "Invalid token")
return fake_db.get_orders(user["id"])
核心问题：代码复用率低、修改成本高、可测试性差。一旦认证规则变化，你需要手动修改每个路由函数。
用依赖注入解决
FastAPI 通过 Depends 让路由声明式地表达自己的需求。你把公共逻辑抽成一个函数，然后让需要的路由“依赖”它：
# ✅ 依赖注入：抽成公共依赖，按需注入
from fastapi import Depends, Header, HTTPException
# 1. 抽成公共认证依赖
async def get_current_user(
authorization: str = Header(...)
) -> dict:
if not authorization.startswith("Bearer "):
raise HTTPException(401, "Invalid token format")
token = authorization.replace("Bearer ", "")
user = fake_db.get_user(token)
if not user:
raise HTTPException(401, "Invalid token")
return user
# 2. 路由只需声明需要的依赖
@app.get("/users/me")
async def get_me(user: dict = Depends(get_current_user)):
return user
@app.get("/orders")
async def get_orders(user: dict = Depends(get_current_user)):
return fake_db.get_orders(user["id"])
现在业务逻辑一目了然，路由只负责返回数据，认证细节全部交给依赖处理。
核心优势
- 代码复用：公共逻辑只需定义一次
- 解耦清晰：路由只关注业务逻辑
- 可测试性强：依赖可轻松替换为模拟对象
- 维护成本低：修改一处即可全局生效
- 内置优化：请求级缓存避免重复执行（后文细讲）
基础用法
1. Depends 函数详解
Depends(可调用对象) 是核心——你可以传入普通函数、异步函数、类甚至生成器。FastAPI 会自动分析依赖的参数并注入相应的请求数据。
# 普通函数依赖：处理查询参数
def get_search_query(q: str = "default"):
return f"搜索内容: {q}"
@app.get("/search")
async def search(q: str = Depends(get_search_query)):
return {"result": q}
当访问 /search?q=fastapi 时，依赖会自动获取 q 的值并传入路由。
2. 带 Query/Path 的依赖工厂
依赖的参数也可以使用 FastAPI 的验证工具（如 Query、Path），构建更强大的依赖工厂：
from fastapi import Query
# 分页依赖工厂
def pagination(
page: int = Query(1, ge=1, description="页码"),
page_size: int = Query(10, ge=1, le=100, description="每页条数")
):
return {"page": page, "page_size": page_size}
@app.get("/items")
async def list_items(p: dict = Depends(pagination)):
# 自动解析并验证参数
return {"data": ["item1", "item2"], "pagination": p}
这样所有需要分页的路由都可以复用这段验证逻辑，同时自动获得了 API 文档中的参数说明。
3. 类作为依赖
当逻辑稍复杂时，可以使用类作为依赖。FastAPI 会自动调用 __init__ 实例化，让你可以封装多个方法和状态：
class AuthService:
def __init__(self, authorization: str = Header(...)):
self.token = authorization.replace("Bearer ", "")
self.user = self._verify()
def _verify(self):
if self.token == "admin123":
return {"id": 1, "role": "admin"}
elif self.token == "user456":
return {"id": 2, "role": "user"}
raise HTTPException(401, "Invalid token")
@app.get("/profile")
async def get_profile(auth: AuthService = Depends()):
# 类实例直接使用，简洁直观
return auth.user
⚠️ 注意：当依赖是类时，Depends() 不需要传入类名，因为 FastAPI 会自行推断。
数据库连接与依赖
依赖注入非常适合管理数据库连接——请求开始时获取会话，请求结束后自动释放。
异步 SQLAlchemy 依赖示例
利用 yield 可以实现资源的优雅管理：
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
# 配置连接池
DATABASE_URL = "sqlite+aiosqlite:///./app.db"
async_engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=30,
pool_pre_ping=True
)
AsyncSessionLocal = sessionmaker(
async_engine, class_=AsyncSession, expire_on_commit=False
)
# 数据库依赖
async def get_db() -> AsyncSession:
async with AsyncSessionLocal() as session:
try:
yield session
except Exception:
await session.rollback()
raise
else:
await session.commit()
# 使用方式
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db)
):
from sqlalchemy import select
from models import User
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(404, "User not found")
return user
请求期间，yield 之前的代码负责创建会话并注入路由；请求结束后，yield 之后的代码负责提交事务或回滚。整个过程对业务代码透明。
依赖链与级联
依赖可以嵌套其他依赖，形成清晰的依赖链，每一层只做好自己的事。
三层认证授权依赖链示例
# 第一层：提取并清理 Token
def get_token(authorization: str = Header(...)):
if not authorization.startswith("Bearer "):
raise HTTPException(401, "Invalid token format")
return authorization.replace("Bearer ", "")
# 第二层：根据 Token 验证用户（依赖第一层）
def get_current_user(token: str = Depends(get_token)):
user = fake_db.get_user(token)
if not user:
raise HTTPException(401, "Invalid token")
return user
# 第三层：检查管理员权限（依赖第二层）
def require_admin(current_user: dict = Depends(get_current_user)):
if current_user["role"] != "admin":
raise HTTPException(403, "Admin access required")
return current_user
# 使用依赖链
@app.get("/admin/dashboard")
async def admin_dashboard(admin: dict = Depends(require_admin)):
return {"admin": admin}
顺序非常直观：require_admin 依赖 get_current_user，而 get_current_user 又依赖 get_token。每一层只处理自己关心的问题，组合起来就构建出强大的权限控制。
可选依赖与默认值
有些场景下，用户可能登录也可能未登录，此时可以通过 Optional 创建可选依赖。
from typing import Optional
# 可选认证依赖
async def get_optional_user(
authorization: Optional[str] = Header(None)
) -> Optional[dict]:
if not authorization:
return None
return fake_db.get_user(authorization.replace("Bearer ", ""))
# 公开接口，有用户信息返回个性化内容，否则返回通用内容
@app.get("/home")
async def home(user: Optional[dict] = Depends(get_optional_user)):
if user:
return {"message": f"Welcome back, {user['name']}!"}
return {"message": "Welcome, guest!"}
当请求携带 Authorization 头时，依赖返回用户对象；否则返回 None。路由根据结果提供不同体验，既灵活又安全。
依赖缓存机制
1. 默认请求级缓存
在同一个请求中，如果多个路由或依赖链都使用了同一个依赖，FastAPI 默认只会执行一次，结果会被缓存复用。这可以避免重复查询数据库或进行重复计算。
def get_user():
print("🔍 查询用户（只打印一次）")
return {"id": 1, "name": "Alice"}
@app.get("/a")
@app.get("/b")
async def endpoints(user: dict = Depends(get_user)):
return {"user": user}
即使 /a 和 /b 被同时调用（或者一个请求中多次调用），你也只会看到一次打印。这就是缓存带来的性能优化。
2. 禁用缓存：use_cache=False
但如果你的依赖需要每次返回不同结果（例如生成 UUID 或读取实时数据），可以通过 use_cache=False 关闭缓存：
import uuid
def get_request_id():
return str(uuid.uuid4())
@app.get("/uuid")
async def get_uuid(
req_id1: str = Depends(get_request_id, use_cache=False),
req_id2: str = Depends(get_request_id, use_cache=False)
):
# req_id1 和 req_id2 会不同
return {"req_id1": req_id1, "req_id2": req_id2}
此时两个参数会分别调用依赖，获得各自独立的 UUID。
实战：认证授权体系
下面我们用依赖注入构建一个可组合、可扩展的认证授权体系。
依赖分层设计
请求 → 中间件（可选） → 依赖注入链
↓
1. get_db （数据库连接）
2. get_token （提取 Token）
3. get_current_user（验证用户）
4. require_roles （角色检查）
↓
路由处理器
核心代码示例
# auth_dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import List
security = HTTPBearer(auto_error=False)
def decode_token(token: str) -> dict:
# 模拟 JWT 验证
if token == "admin":
return {"id": 1, "role": "admin", "sub": "admin@example.com"}
elif token == "editor":
return {"id": 2, "role": "editor", "sub": "editor@example.com"}
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid token")
def require_roles(allowed_roles: List[str]):
"""角色权限检查工厂函数"""
async def checker(
cred: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
user = decode_token(cred.credentials)
if user["role"] not in allowed_roles:
raise HTTPException(status.HTTP_403_FORBIDDEN, "Permission denied")
return user
return checker
# 预定义常用角色依赖
require_admin = require_roles(["admin"])
require_editor = require_roles(["admin", "editor"])
# main.py
from fastapi import FastAPI, Depends
from auth_dependencies import require_admin, require_editor
app = FastAPI()
@app.get("/public")
async def public():
return {"status": "ok"}
@app.get("/me")
async def me(user: dict = Depends(require_editor)):
return user
@app.delete("/users/{user_id}")
async def delete_user(user_id: int, _: dict = Depends(require_admin)):
# 下划线表示不需要使用返回值
return {"deleted": user_id}
这种设计让你可以像搭积木一样组合权限：只需要改变 require_roles 的参数，就能复用同一套认证逻辑。
进阶用法
1. 全局路由级依赖
在 APIRouter 或 include_router 中添加 dependencies，可以实现一组路由共用前置依赖，非常适合为整个模块添加权限保护：
from fastapi import APIRouter
admin_router = APIRouter(
prefix="/admin",
dependencies=[Depends(require_admin)] # 所有 admin 路由都需要管理员权限
)
@admin_router.get("/users")
async def list_admin_users():
return {"users": ["admin1"]}
app.include_router(admin_router)
在这里，你不需要在每个路由中重复添加依赖，整个子模块的访问控制由框架统一管理。
避坑指南与性能优化
避坑指南
- 避免副作用：依赖函数应保持幂等，尽量不修改全局状态。
- 避免循环依赖：不要让 A 依赖 B，而 B 又依赖 A。
- 避免过度嵌套：依赖链深度一般不超过 3-4 层，否则调试困难。
- 避免异步依赖中的阻塞操作：如果依赖中有 IO 密集型或 CPU 密集型代码，应使用
run_in_threadpool 或相关工具移至线程池。
性能优化
- 善用默认请求级缓存：减少重复的数据库查询或计算。
- 配置合理的数据库连接池：避免连接耗尽，如上文中
pool_size 和 max_overflow 的设置。
- 依赖执行顺序优化：把最可能快速失败的验证放在依赖链前面（如 Token 格式检查），让请求尽早返回，节省资源。
总结
💡 核心思想：依赖注入让代码变得可测试、可复用、声明式、易维护，是构建企业级 FastAPI 应用的核心工具。
🔗 官方扩展阅读
