在上一篇教程中,我们学习了 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-urlencodedmultipart/form-data 格式发送。FastAPI 通过 FormFile 参数支持这些格式。

2.1 安装 python-multipart

要处理表单数据,需要安装 python-multipart

pip install python-multipart

2.2 使用 Form 参数接收表单数据

使用 fastapi.Form 声明表单字段,类似于 QueryBody

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
  • 注意:你不能在同一个路径操作中同时声明 FormBody 参数(来自 JSON),因为请求只能有一种主体格式。但你可以混合使用 FormFilemultipart/form-data)。

2.4 文件上传(File)

文件上传使用 fastapi.Filefastapi.UploadFileUploadFile 提供了更高级的接口,可以直接读取文件内容。

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 混合表单字段和文件

表单中可以同时包含文本字段和文件字段。只需在路径操作函数中混合使用 FormFile

@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 文件上传(FileUploadFile),包括混合表单字段和文件。
  • 其他高级主题:简要介绍了中间件、CORS、后台任务和测试,这些在实际项目中非常实用。

通过这两篇教程的学习,你应该已经掌握了 FastAPI 的核心功能和常见用法,能够独立构建功能完善的 Web API。Happy coding!

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐