- Published on
FastAPI 依赖注入机制详解及
- Authors
- Name
- 不作声
- GitHub
- Github @buzuosheng
本文发布于 2025-01-01,主要介绍 FastAPI 的依赖注入机制及其实现原理,并与 Nest.js 进行对比分析。
FastAPI 依赖注入机制详解
FastAPI 是一个现代、高性能的 Python Web 框架,专注于构建 API。它的依赖注入系统不仅功能强大,还与 Python 的类型提示系统完美结合,让我们能够编写出更加模块化、可测试和可维护的代码。
一、依赖注入基础
1. 依赖注入的概念
依赖注入(Dependency Injection,简称 DI)是一种软件设计模式,它的核心思想是将依赖关系的创建与业务逻辑分离。在 FastAPI 中,依赖注入通过 Depends
类实现,它能够:
- 自动处理依赖的生命周期
- 提供类型安全的依赖管理
- 支持异步操作
- 简化测试和模拟
2. FastAPI 依赖注入的基本用法
以数据库连接为例,展示 FastAPI 依赖注入的基本使用方式:
from fastapi import Depends, FastAPI
from typing import Generator
app = FastAPI()
async def get_db() -> Generator:
db = Database()
try:
yield db # 使用 yield 实现上下文管理
finally:
await db.close() # 确保资源正确释放
@app.get("/users/")
async def read_users(db: Database = Depends(get_db)):
return await db.get_all_users()
3. FastAPI 依赖注入的高级特性
a) 参数化依赖
FastAPI 支持创建可配置的依赖项:
from typing import Optional
def pagination(
page: Optional[int] = 1,
page_size: Optional[int] = 10,
max_page_size: int = 100
) -> dict:
# 参数验证和处理
final_size = min(page_size, max_page_size)
skip = (page - 1) * final_size
return {
"skip": skip,
"limit": final_size,
"page": page
}
@app.get("/items/")
async def list_items(
pagination_params: dict = Depends(pagination)
):
return {
"params": pagination_params,
"items": [] # 实际项目中从数据库获取数据
}
b) 依赖嵌套与链式依赖
FastAPI 支持复杂的依赖关系链:
from fastapi import Depends, HTTPException
from typing import Optional
async def get_current_user(
token: str,
db: Database = Depends(get_db)
) -> Optional[User]:
user = await db.get_user_by_token(token)
if not user:
raise HTTPException(status_code=401)
return user
async def get_admin_user(
current_user: User = Depends(get_current_user)
) -> User:
if not current_user.is_admin:
raise HTTPException(status_code=403)
return current_user
@app.get("/admin/dashboard")
async def admin_dashboard(
admin: User = Depends(get_admin_user)
):
return {"message": f"Welcome admin {admin.username}"}
c) 基于 Pydantic 的依赖验证
结合 Pydantic 模型,实现更强大的参数验证:
from pydantic import BaseModel, Field
from typing import Optional
class PaginationParams(BaseModel):
page: int = Field(default=1, ge=1)
page_size: int = Field(default=10, ge=1, le=100)
sort_by: Optional[str] = None
order: Optional[str] = Field(
default="asc",
pattern="^(asc|desc)$"
)
@app.get("/users/")
async def list_users(
params: PaginationParams, # FastAPI 自动处理参数验证
db: Database = Depends(get_db)
):
users = await db.get_users(
skip=(params.page - 1) * params.page_size,
limit=params.page_size,
sort_by=params.sort_by,
order=params.order
)
return {
"pagination": params.dict(),
"data": users
}
这种方式的优势:
- 自动参数验证和类型转换
- 完整的 OpenAPI 文档生成
- IDE 智能提示支持
- 清晰的错误提示
- 易于测试和维护
二、最佳实践建议
依赖函数命名:使用描述性的名称,如
get_current_user
而不是user
异常处理:在依赖函数中妥善处理异常,返回合适的 HTTP 状态码
资源管理:使用
yield
语句确保资源正确释放类型提示:始终添加类型注解,提高代码可维护性
依赖缓存:合理使用
use_cache=False
参数控制依赖缓存行为
@app.get("/items/")
async def read_items(
fresh_db: Database = Depends(get_db, use_cache=False)
):
return await fresh_db.get_items()
通过合理使用 FastAPI 的依赖注入系统,我们可以构建出更加模块化、可测试和可维护的应用程序。