Python之FastAPI 开发框架(第二篇):深入核心概念与高级功能
在上一篇教程中,我们学习了 FastAPI 的基础知识:安装、第一个应用、自动文档、路由、请求/响应处理以及 Pydantic 模型的使用。掌握了这些基础后,你已经可以构建简单的 API 了。然而,在实际开发中,我们经常需要处理更复杂的场景,比如依赖注入、表单数据、文件上传、中间件等。本篇教程将深入探讨这些核心概念和高级功能,帮助你进一步提升 FastAPI 开发能力。
1. 路径操作依赖项
依赖项(Dependency Injection)是 FastAPI 最强大的特性之一。它允许你在路径操作函数中声明一些在运行前需要“解决”的依赖,例如:获取当前用户、验证 API 密钥、记录请求日志、数据库会话管理等。依赖项可以是函数、类,甚至可以是其他依赖项。
1.1 什么是依赖项
简单来说,依赖项是一个可调用的对象(函数、类等),它可以接收参数,并返回一个值。FastAPI 会在调用路径操作函数之前调用这个依赖项,并将其返回值注入到路径操作函数的参数中。
依赖项通常用于:
- 共享通用逻辑(如认证、数据库连接)
- 代码复用
- 提高可测试性
1.2 定义依赖项
最简单的依赖项是一个普通的 Python 函数:
from fastapi import Depends, FastAPI
app = FastAPI()
# 定义一个依赖项函数
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
def read_users(commons: dict = Depends(common_parameters)):
return commons
Depends(common_parameters)告诉 FastAPI 在调用read_items之前先调用common_parameters函数,并将其返回值作为commons参数传递给read_items。- 依赖项函数可以像路径操作函数一样声明参数(查询参数、路径参数、请求体等),FastAPI 会自动处理这些参数。
1.3 类作为依赖项
除了函数,你也可以使用类作为依赖项。FastAPI 会调用类的 __init__ 方法(如果定义了)或者直接将类实例化。
class CommonQueryParams:
def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
FastAPI 也支持简写:commons: CommonQueryParams = Depends(),它会自动推断依赖项为 CommonQueryParams。
1.4 依赖项的共享和嵌套
依赖项可以依赖于其他依赖项,形成依赖树。FastAPI 会确保每个依赖项只被调用一次(在同一个请求中)。
def get_db():
# 模拟数据库连接
return "db_session"
def get_current_user(db=Depends(get_db)):
# 模拟获取当前用户
return {"username": "testuser", "db": db}
@app.get("/user/")
def read_user(current_user: dict = Depends(get_current_user)):
return current_user
1.5 全局依赖
你可以将依赖项应用于整个应用或一组路径操作。例如,对所有路径进行 API 密钥验证。
async def verify_token(x_token: str = Header(...)):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
return x_token
app = FastAPI(dependencies=[Depends(verify_token)])
现在所有路径操作都会先调用 verify_token。
1.6 带参数的依赖项
有时依赖项本身需要参数(例如,权限级别)。你可以创建一个返回依赖项的函数(即“依赖项工厂”)。
def require_permission(permission: str):
def permission_dependency(current_user: dict = Depends(get_current_user)):
if permission not in current_user.get("permissions", []):
raise HTTPException(status_code=403, detail="Permission denied")
return current_user
return permission_dependency
@app.get("/admin/")
def admin(user: dict = Depends(require_permission("admin"))):
return {"message": "Welcome admin"}
这里 require_permission("admin") 返回一个内部函数,该内部函数作为实际的依赖项,它依赖于 get_current_user,并检查权限。
1.7 依赖项的 yield(上下文管理器)
FastAPI 支持使用 yield 的依赖项,用于需要在请求后执行清理操作的场景,例如关闭数据库连接。这种依赖项必须配合 @contextlib.contextmanager 或 @asynccontextmanager 使用。
from contextlib import asynccontextmanager
@asynccontextmanager
async def get_db_session():
db = DBSession()
try:
yield db
finally:
db.close()
@app.get("/items/")
def read_items(db=Depends(get_db_session)):
# 使用 db
return db.query(...)
在 yield 之前的代码在请求前执行,yield 之后(finally 块)在响应发送后执行。
2. 表单数据
处理 HTML 表单提交时,数据通常以 application/x-www-form-urlencoded 或 multipart/form-data 格式发送。FastAPI 通过 Form 和 File 参数支持这些格式。
2.1 安装 python-multipart
要处理表单数据,需要安装 python-multipart:
pip install python-multipart
2.2 使用 Form 参数接收表单数据
使用 fastapi.Form 声明表单字段,类似于 Query 和 Body。
from fastapi import FastAPI, Form
app = FastAPI()
@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
return {"username": username}
Form(...)表示该字段是必填的。- 表单数据通常用于简单的键值对,不支持嵌套结构(与 JSON 不同)。
2.3 表单与 JSON 的区别
- JSON 数据:使用 Pydantic 模型,内容类型为
application/json。 - 表单数据:使用
Form参数,内容类型为application/x-www-form-urlencoded。 - 注意:你不能在同一个路径操作中同时声明
Form和Body参数(来自 JSON),因为请求只能有一种主体格式。但你可以混合使用Form和File(multipart/form-data)。
2.4 文件上传(File)
文件上传使用 fastapi.File 和 fastapi.UploadFile。UploadFile 提供了更高级的接口,可以直接读取文件内容。
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/files/")
async def upload_file(file: bytes = File(...)):
# 将文件内容作为字节读取,适合小文件
return {"file_size": len(file)}
@app.post("/uploadfile/")
async def upload_uploadfile(file: UploadFile = File(...)):
# UploadFile 提供了 filename、content_type 等属性,以及异步读取方法
contents = await file.read()
return {"filename": file.filename, "content_type": file.content_type}
UploadFile是推荐的方式,因为它不会将所有内容加载到内存中(除非调用read()),适合大文件。- 使用
async方法读取文件内容:await file.read(),await file.write()等。
2.5 多个文件上传
可以同时接收多个文件,使用 List[UploadFile]。
from typing import List
@app.post("/multiple-files/")
async def upload_multiple_files(files: List[UploadFile] = File(...)):
return [{"filename": f.filename} for f in files]
2.6 混合表单字段和文件
表单中可以同时包含文本字段和文件字段。只需在路径操作函数中混合使用 Form 和 File。
@app.post("/user/")
async def create_user(
username: str = Form(...),
profile_picture: UploadFile = File(...)
):
return {"username": username, "profile_picture": profile_picture.filename}
FastAPI 会自动将请求解析为 multipart/form-data。
3. 其他高级主题(选学)
除了依赖项和表单数据,FastAPI 还提供了许多实用的高级功能。以下是一些常用功能的简要介绍,帮助你进一步扩展 API 能力。
3.1 中间件
中间件允许你在每个请求处理前后执行代码。可以用于添加自定义请求头、日志记录、性能分析等。
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
3.2 CORS(跨域资源共享)
如果你的 API 需要被浏览器中的前端应用(如 React、Vue)访问,必须配置 CORS。FastAPI 提供了 CORSMiddleware 来轻松处理。
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
"http://localhost",
"http://localhost:8080",
"https://myfrontend.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
3.3 后台任务
你可以将一些需要在响应返回后执行的操作(如发送邮件、处理数据)放到后台任务中。
from fastapi import BackgroundTasks
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
@app.post("/send-notification/")
async def send_notification(background_tasks: BackgroundTasks, email: str):
background_tasks.add_task(write_log, f"Notification sent to {email}\n")
return {"message": "Notification sent in background"}
3.4 测试 FastAPI 应用
FastAPI 基于 Starlette,因此你可以使用 TestClient(来自 starlette.testclient)进行测试。推荐使用 pytest。
from fastapi.testclient import TestClient
from .main import app
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, FastAPI!"}
安装 pytest 后,运行 pytest 即可执行测试。
4. 总结
在第二篇教程中,我们深入探讨了以下核心概念:
- 路径操作依赖项:如何通过
Depends实现依赖注入,共享通用逻辑,处理认证、数据库会话等。学习了函数依赖、类依赖、嵌套依赖、全局依赖以及带参数的依赖项。 - 表单数据:如何接收
application/x-www-form-urlencoded表单数据(Form)和multipart/form-data文件上传(File、UploadFile),包括混合表单字段和文件。 - 其他高级主题:简要介绍了中间件、CORS、后台任务和测试,这些在实际项目中非常实用。
通过这两篇教程的学习,你应该已经掌握了 FastAPI 的核心功能和常见用法,能够独立构建功能完善的 Web API。Happy coding!
更多推荐



所有评论(0)