激流快艇苹果版
29.1M · 2025-09-12
依赖注入(Dependency Injection)是一种设计模式,通过外部提供组件所需的依赖,避免组件自行创建或管理依赖。在 FastAPI 中,依赖注入用于管理路由函数所需的资源(如数据库连接、配置文件等),使代码更模块化、可测试性强。
类比: 想象一家餐厅(路由函数),顾客(请求)需要食物(依赖)。依赖注入相当于中央厨房(FastAPI 系统)统一配送食材,无需餐厅自己种菜或养殖。
FastAPI 的依赖注入系统基于 Python 的类型提示和 Depends
函数。工作流程如下:
flowchart TD
A[请求路由] --> B[解析依赖项]
B --> C{依赖项是否被覆盖?}
C -- 是 --> D[使用覆盖的依赖]
C -- 否 --> E[使用默认依赖]
D & E --> F[执行路由逻辑]
目的: 在测试或特殊场景中替换默认依赖(如用虚拟数据库代替真实数据库)。
覆盖方法:
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
# 默认依赖
def get_db():
return "Real Database Connection"
@app.get("/items")
def read_items(db: str = Depends(get_db)):
return {"db": db}
# 测试时覆盖依赖
def override_get_db():
return "Mock Database Connection"
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
response = client.get("/items")
print(response.json()) # 输出: {"db": "Mock Database Connection"}
关键点:
dependency_overrides
是全局字典,键为原依赖函数,值为覆盖函数。from fastapi import Depends, FastAPI
from pydantic import BaseModel
app = FastAPI()
# 默认验证逻辑
def verify_token(token: str):
if token != "valid_token":
raise ValueError("Invalid Token")
return True
# 覆盖逻辑:总返回验证成功
def mock_verify_token(token: str):
return True
# 测试场景
app.dependency_overrides[verify_token] = mock_verify_token
@app.post("/secure")
def secure_endpoint(verified: bool = Depends(verify_token)):
return {"status": "access granted"}
# 测试时传递无效 token 也不会报错
client = TestClient(app)
response = client.post("/secure", headers={"token": "invalid_token"})
assert response.json()["status"] == "access granted"
问题: 422 Validation Error
原因: 覆盖函数与原函数签名不一致(参数数量/类型不同)。
解决方案:
class Token(BaseModel):
token: str
def mock_verify_token(token: Token): # 与原函数参数模型一致
return True
应用通常需要在不同环境(开发/测试/生产)中使用不同配置,例如:
flowchart TD
A[启动应用] --> B[读取环境变量]
B --> C{环境类型}
C -- 开发 --> D[加载开发配置]
C -- 测试 --> E[加载测试配置]
C -- 生产 --> F[加载生产配置]
D & E & F --> G[注入路由依赖]
步骤 1: 定义环境配置模型
from pydantic import BaseSettings
class Settings(BaseSettings):
env: str = "dev" # 默认开发环境
db_url: str = ""
class Config:
env_file = ".env" # 从 .env 文件加载配置
settings = Settings()
步骤 2: 按环境切换依赖
from fastapi import Depends
def get_db(settings: Settings = Depends()):
if settings.env == "dev":
return "sqlite:///dev.db"
elif settings.env == "prod":
return "postgresql://user:[email protected]"
else:
return "mock://test.db"
@app.get("/data")
def fetch_data(db_url: str = Depends(get_db)):
return {"db_url": db_url}
优化方案: 避免 if-else
,使用注册表模式
_db_registry = {
"dev": "sqlite:///dev.db",
"test": "mock://test.db",
"prod": "postgresql://user:[email protected]"
}
def get_db(settings: Settings = Depends()):
return _db_registry[settings.env] # 直接映射,避免分支判断
敏感数据管理:
.env
文件存储密钥,将其加入 .gitignore
。SecretStr
类型隐藏敏感字段:class Settings(BaseSettings):
api_key: SecretStr
环境变量验证:
from pydantic import validator
class Settings(BaseSettings):
env: str
@validator("env")
def validate_env(cls, v):
if v not in ["dev", "test", "prod"]:
raise ValueError("Invalid environment")
return v
问题: 如何在 FastAPI 中临时覆盖一个依赖项?
答案: 使用 app.dependency_overrides[original_dep] = mock_dep
。覆盖需保证函数签名一致。
问题: 为什么在多环境配置中推荐使用 Pydantic 的 BaseSettings
?
答案: 它自动从环境变量或 .env
文件加载配置,支持类型验证和默认值,避免手动解析。
fastapi == 0.111.0
pydantic == 2.7.1
python-dotenv == 1.0.1 # 用于加载 .env 文件