你是否曾盯着 FastAPI 的代码发呆,疑惑它是如何“魔法般”地处理所有函数参数的?你并不孤单!FastAPI 的参数处理系统非常强大,但刚开始时它的工作方式可能让人觉得像是在解读外星科技。
泡上一杯咖啡 ☕️,让我们彻底揭开 FastAPI 参数的神秘面纱。
路径参数:来自 URL 的值
让我们从最简单的情况开始。路径参数直接来自 URL:
1
2
3
|
@app.get("/users/{user_id}")
def read_user(user_id: int):
return {"user_id": user_id}
|
当有人访问 /users/42
时,FastAPI 会从 URL 中提取 42
,并作为 user_id
传递给你的函数。类型注解(int
)会让 FastAPI 自动将 URL 中的字符串转换为整数。
查询参数:来自 ?key=value
查询参数来自 URL 的查询字符串(?
后面):
1
2
3
|
@app.get("/search/")
def search(q: str, page: int = 1, limit: int = 10):
return {"query": q, "page": page, "limit": limit}
|
请求 /search/?q=fastapi&limit=5
时,你会得到 {"query": "fastapi", "page": 1, "limit": 5}
。带有默认值的参数(如 page
和 limit
)是可选的。
请求体:JSON 数据
对于 POST、PUT 等需要发送数据的方法,你通常会使用 Pydantic 模型来处理 JSON:
1
2
3
4
5
6
7
8
9
10
|
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
is_offer: bool = False
@app.post("/items/")
def create_item(item: Item):
return {"item_name": item.name, "price_with_tax": item.price * 1.1}
|
当你的 API 收到如 {"name": "Coffee Mug", "price": 12.99}
的 JSON 时,FastAPI 会根据你的模型进行校验,并创建一个合适的 Python 对象。
表单数据与文件上传
用于处理表单提交和文件上传:
1
2
3
4
5
6
7
8
9
10
11
|
from fastapi import File, Form, UploadFile
@app.post("/upload/")
async def upload_file(
name: str = Form(...),
file: UploadFile = File(...)
):
return {
"filename": file.filename,
"name": name
}
|
Form(...)
和 File(...)
告诉 FastAPI 这些值应从表单数据中获取,而不是 JSON。
依赖注入的魔法
这正是 FastAPI 闪耀之处。通过 Depends()
函数,你可以注入数据库连接、认证等:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
def get_db():
db = SessionLocal()
try:
yield db # yield 很重要!
finally:
db.close() # 这会在你的端点函数执行后运行
@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
|
当 FastAPI 看到 Depends(get_db)
时,它会:
- 调用
get_db()
函数
- 获取 yield 出来的对象(数据库会话)
- 将其传递给你的端点函数
- 在你的函数完成后,继续执行
get_db()
以运行清理代码
这种模式非常适合资源管理——如连接、文件等。
依赖注入实现认证
依赖注入的一个常见用途是认证:
1
2
3
4
5
6
7
8
|
def get_current_user(token: str = Header(...)):
if token != "secret-token":
raise HTTPException(status_code=401, detail="Invalid token")
return {"username": "john_doe"}
@app.get("/users/me")
def read_user_me(current_user: dict = Depends(get_current_user)):
return current_user
|
这里,get_current_user
会检查 token,然后:
- 返回一个用户对象,传递给你的函数
- 或抛出异常,阻止你的函数运行
FastAPI 的特殊参数(无需 Depends)
FastAPI 能自动识别某些类型并自动注入,无需 Depends()
:
1
2
3
4
5
6
7
8
9
10
11
|
from fastapi import Request, Response, BackgroundTasks
@app.get("/special")
async def special_parameters(
request: Request, # 原始 HTTP 请求
response: Response, # 可修改的 HTTP 响应
background_tasks: BackgroundTasks, # 用于响应后处理
):
response.set_cookie("visited", "true")
background_tasks.add_task(log_visit, request.client.host)
return {"message": "Check your cookies!"}
|
当 FastAPI 看到参数类型为 Request
、Response
或 BackgroundTasks
时,会自动提供相应对象。这就是为什么在你的示例中,background_tasks: BackgroundTasks
不需要 Depends()
——它是特殊的!
理解 BackgroundTasks
BackgroundTasks
值得特别关注,因为它非常实用:
1
2
3
4
5
6
7
8
9
10
11
12
|
def process_item(item_id: int):
# 这在响应发送后运行
print(f"Processing item {item_id}")
# 可以进行数据库操作、发送邮件等
@app.post("/items/{item_id}/process")
async def process_item_endpoint(
item_id: int,
background_tasks: BackgroundTasks
):
background_tasks.add_task(process_item, item_id)
return {"message": "Processing started"}
|
任务会在响应发送后运行,让 API 响应更快,而耗时操作在后台进行。
依赖链:依赖的依赖
依赖可以有自己的依赖,形成依赖链:
1
2
3
4
5
6
7
8
9
10
11
12
13
|
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_repository(db: Session = Depends(get_db)):
return Repository(db)
@app.get("/items/")
def read_items(repo: Repository = Depends(get_repository)):
return repo.get_all_items()
|
FastAPI 会自动解析整个依赖链:
- 调用
get_db()
获取数据库会话
- 将其传递给
get_repository()
获取仓库对象
- 再将仓库对象传递给你的端点
依赖缓存:共享依赖只调用一次
如果你在同一个请求中多次使用同一个依赖,FastAPI 足够智能,只会调用一次:
1
2
3
4
5
6
7
8
9
10
11
|
def get_user(user_id: int, db: Session = Depends(get_db)):
return db.query(User).filter(User.id == user_id).first()
@app.get("/items/{item_id}")
def read_item(
item_id: int,
user: User = Depends(get_user),
db: Session = Depends(get_db) # 和 get_user 里用的是同一个 db 会话!
):
item = db.query(Item).filter(Item.id == item_id).first()
return {"item": item, "owner": user}
|
即使 get_db
被直接和间接(在 get_user
里)调用,FastAPI 也只会调用一次。
测试中的依赖覆盖
依赖注入最强大的功能之一,就是让测试变得非常简单:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
# 在你的测试文件中
from fastapi.testclient import TestClient
from app.main import app, get_db
def get_test_db():
test_db = TestingSessionLocal()
try:
yield test_db
finally:
test_db.close()
app.dependency_overrides[get_db] = get_test_db
client = TestClient(app)
def test_read_item():
response = client.get("/items/1")
assert response.status_code == 200
|
通过用 get_test_db
替换 get_db
,你的所有端点都会使用测试数据库。
Annotated 优雅参数类型(FastAPI 0.95.0+)
为了让代码更简洁,FastAPI 现在支持 typing
模块中的 Annotated
:
1
2
3
4
5
6
7
8
9
10
11
12
13
|
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
def get_user_agent(user_agent: str = Header(None)):
return user_agent
@app.get("/ua")
async def read_user_agent(
user_agent: Annotated[str, Depends(get_user_agent)]
):
return {"user_agent": user_agent}
|
这样可以将类型信息和依赖信息分离,让代码更易读。
综合示例:多种参数类型的结合
下面是一个结合多种参数类型的完整示例:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
import uuid
from fastapi import BackgroundTasks, Depends, FastAPI, File, UploadFile
from sqlalchemy.orm import Session
app = FastAPI()
@app.post("/{case_id}/documents/upload", status_code=202)
async def upload_documents(
case_id: uuid.UUID, # 来自 URL 路径
background_tasks: BackgroundTasks, # FastAPI 自动注入(服务端)
file: UploadFile = File(...), # 来自客户端(multipart/form-data)
db: Session = Depends(get_db), # FastAPI 自动注入(服务端)
current_user: User = Depends(get_current_active_user), # FastAPI 自动注入(服务端)
):
# 将文件信息存储到数据库
document = Document(
case_id=case_id,
filename=file.filename,
uploaded_by=current_user.id
)
db.add(document)
db.commit()
# 在后台处理文件
background_tasks.add_task(
process_document,
document_id=document.id,
file=file,
user_id=current_user.id
)
return {"message": "Document upload accepted for processing"}
|
这个端点优雅地处理了:
- 路径参数(
case_id
)
- 文件上传(
file
)
- 数据库连接(
db
)
- 认证(
current_user
)
- 后台处理(
background_tasks
)
何时使用哪种参数类型
总结一下各参数类型的使用场景:
- 路径参数:用于 URL 中必须的值
- 查询参数:用于可选的过滤、排序、分页
- 请求体:用于 POST/PUT 请求中的复杂数据
- 表单数据:用于传统 HTML 表单
- 文件上传:用于文件处理
- 依赖:用于共享资源、认证和业务逻辑
- 特殊类型:用于直接访问请求/响应对象和后台任务
总结
FastAPI 的参数系统一开始可能看起来很复杂,但理解之后你会发现它极大简化了 API 开发。它自动处理从 URL 提取到依赖注入的一切,让你专注于业务逻辑,而不是样板代码。
记住基本规则:
- 路径和查询参数来自 URL
- 请求体来自请求数据
- 依赖和特殊类型由 FastAPI 提供
掌握这些概念后,你很快就能构建出简洁、可维护的 API。祝你编码愉快!🚀