使用 FastAPI 快速开发 API：从入门到精通
引言：为什么选择 FastAPI？
在当今快节奏的软件开发领域，构建高性能、易于维护且能够快速迭代的 API 是成功的关键。Python 作为一门广泛应用于后端开发的语言，拥有众多成熟的 Web 框架。从老牌的 Django 和 Flask，到后起之秀 FastAPI，开发者面临着丰富的选择。
FastAPI 是一个相对年轻但发展迅猛的 Python Web 框架。它的出现，极大地提升了 Python 开发 API 的效率和性能。FastAPI 基于标准的 Python 类型提示构建，集成了 Starlette（高性能 ASGI 框架）和 Pydantic（数据验证和序列化库），从而提供了一系列令人惊叹的特性：
- 极高的性能：得益于 Starlette 和 Uvicorn (或其他 ASGI 服务器)，FastAPI 拥有与 NodeJS 和 Go 比肩的性能表现。
- 开发效率高：使用现代 Python 特性，代码简洁易懂。
- 自动交互式 API 文档：基于 OpenAPI 标准，自动生成 Swagger UI 和 ReDoc 文档，极大地简化了 API 接口的联调和管理。
- 自动数据验证和序列化：利用 Pydantic，通过简单的类型提示就能实现请求数据的自动验证和响应数据的自动序列化，有效减少了手动编写验证代码的工作量。
- 强大的依赖注入系统：灵活且易于使用的依赖注入机制，使得代码更易于组织、复用和测试。
- 类型提示支持：充分利用 Python 类型提示，编辑器支持良好，减少运行时错误。
- ASGI 支持：支持异步请求，能够处理高并发。
相比于 Flask 的灵活性但需要手动处理验证、序列化、文档等，以及 Django 的全能但相对笨重且异步支持相对较晚，FastAPI 在构建现代 API 方面展现出了独特的优势，特别适合用于构建微服务、RESTful API 等。
本文将带领你深入了解 FastAPI，从基础概念到核心特性，通过详细的代码示例，让你能够快速掌握使用 FastAPI 开发高性能 API 的技能。
第一步：环境搭建和你的第一个 FastAPI 应用
开始使用 FastAPI 前，你需要安装 Python 3.7+ 版本。然后，使用 pip 安装 FastAPI 和 ASGI 服务器 (推荐 Uvicorn)。
bash
pip install fastapi uvicorn[standard]
uvicorn[standard] 会安装 Uvicorn 的标准版本，包含了一些常用的可选依赖，例如用于热重载的 watchfiles。
接下来，我们创建一个简单的 FastAPI 应用。创建一个名为 main.py 的文件：
“`python
main.py
from fastapi import FastAPI
创建一个 FastAPI 应用实例
app = FastAPI()
定义一个根路径的 GET 请求处理函数
@app.get(“/”)
async def read_root():
“””
根路径接口，返回一个简单的问候消息
“””
return {“message”: “Hello, World!”}
定义另一个路径的 GET 请求处理函数
@app.get(“/items/{item_id}”)
async def read_item(item_id: int, q: str | None = None):
“””
获取指定ID的物品信息
Args:
item_id: 物品的唯一标识符 (整型)
q: 可选的查询参数 (字符串或None)
Returns:
包含物品ID和查询参数的字典
"""
# 这是一个异步函数，虽然这里没有实际的异步操作，
# 但展示了使用 async def 的方式
if q:
return {"item_id": item_id, "q": q}
return {"item_id": item_id}
“`
这是一个非常简单的 FastAPI 应用：
* from fastapi import FastAPI：导入 FastAPI 类。
* app = FastAPI()：创建一个 FastAPI 应用实例。所有接口都将注册到这个实例上。
* @app.get("/")：这是一个装饰器，用于将下面的异步函数 read_root 注册为处理根路径 (/) 的 GET 请求的处理函数（或称为“路径操作函数”）。
* async def read_root():：定义一个异步函数。FastAPI 推荐使用 async def 来定义路径操作函数，特别是在函数内部有 await 操作（如数据库查询、网络请求）时。即使函数是同步的，使用 async def 也可以在某些情况下获得更好的性能，尽管同步函数 (def) 也是支持的，FastAPI 会在线程池中运行它们。
* return {"message": "Hello, World!"}：路径操作函数返回的数据可以是 Python 字典、列表、Pydantic 模型等。FastAPI 会自动将这些数据序列化为 JSON 格式作为响应。
要运行这个应用，打开终端，进入到 main.py 文件所在的目录，然后执行：
bash
uvicorn main:app --reload
main:app：指定应用实例的位置。main是文件名 (模块)，app是文件中创建的 FastAPI 实例的变量名。--reload：启用热重载。当你修改main.py文件并保存时，Uvicorn 会自动重启服务器，非常方便开发。
现在，打开你的浏览器，访问 http://127.0.0.1:8000/，你会看到 {"message": "Hello, World!"}。
访问 http://127.0.0.1:8000/items/5，你会看到 {"item_id": 5}。
访问 http://127.0.0.1:8000/items/5?q=somequery，你会看到 {"item_id": 5, "q": "somequery"}。
注意到 /items/{item_id} 这个路径，{item_id} 表示这是一个路径参数。FastAPI 会自动从 URL 中提取这个参数，并尝试将其转换为函数参数中指定的类型 (int)。如果在访问 /items/abc 时，abc 无法转换为整数，FastAPI 会自动返回一个验证错误响应。这就是 FastAPI 自动数据验证的一个简单体现。同时，q: str | None = None 表示 q 是一个查询参数，类型是字符串或 None，默认值为 None，这意味着它是可选的。
第二步：自动文档和数据验证的魔力
FastAPI 最引人注目的特性之一是其自动生成的 API 文档。启动应用后，访问：
http://127.0.0.1:8000/docs：你会看到交互式的 Swagger UI 文档。http://127.0.0.1:8000/redoc：你会看到 ReDoc 文档。
这些文档是根据你的代码（特别是类型提示、Pydantic 模型和函数文档字符串）自动生成的。你可以直接在 Swagger UI 中测试 API 接口，查看请求参数、响应结构、状态码等信息，这极大地提升了前后端协作的效率。
使用 Pydantic 定义请求体
对于 POST、PUT 等请求，数据通常通过请求体发送，而不是路径或查询参数。FastAPI 使用 Pydantic 来定义请求体的结构和进行数据验证。
安装 Pydantic (如果在安装 uvicorn[standard] 时没有包含):
bash
pip install pydantic
修改 main.py：
“`python
main.py
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional # 明确导入Optional
app = FastAPI()
定义一个 Pydantic 模型来描述请求体的结构
class Item(BaseModel):
name: str
description: Optional[str] = None # 描述是可选的，默认为 None
price: float
tax: Optional[float] = None
定义另一个 Pydantic 模型用于更新
class ItemUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
price: Optional[float] = None
tax: Optional[float] = None
@app.get(“/”)
async def read_root():
return {“message”: “Hello, World!”}
@app.get(“/items/{item_id}”)
async def read_item(item_id: int, q: str | None = None):
if q:
return {“item_id”: item_id, “q”: q}
return {“item_id”: item_id}
定义一个 POST 请求处理函数，接收 Item 模型作为请求体
@app.post(“/items/”)
async def create_item(item: Item):
“””
创建一个新的物品
“””
# item 参数的类型是 Item 模型，FastAPI/Pydantic 会自动验证请求体数据
# 如果验证通过，item 就是一个 Item 实例
# 你可以直接访问 item 的属性，如 item.name, item.price 等
item_dict = item.model_dump() # 将 Pydantic 模型转换为字典 (Pydantic v2+)
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({“price_with_tax”: price_with_tax})
return item_dict # FastAPI 会将字典自动序列化为 JSON
定义一个 PUT 请求处理函数，接收路径参数和请求体
@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: ItemUpdate):
“””
更新指定ID的物品信息
“””
results = {“item_id”: item_id, “item”: item.model_dump()}
return results
“`
在这个例子中：
* 我们定义了一个 Item 类，它继承自 pydantic.BaseModel。这个类定义了我们期望接收的请求体的结构：包含 name (字符串), description (可选字符串), price (浮点数), tax (可选浮点数)。
* 在 @app.post("/items/") 路径操作函数 create_item 中，我们将参数 item 的类型指定为 Item 模型 (item: Item)。FastAPI 会自动：
* 读取请求体的内容（通常是 JSON）。
* 使用 Item 模型对请求体数据进行验证。
* 如果数据无效（例如，缺少必需的字段 name 或 price，或者字段类型不匹配），FastAPI 会自动返回一个 HTTP 422 Unprocessable Entity 响应，其中包含详细的验证错误信息。
* 如果验证通过，FastAPI 会创建一个 Item 实例，并将其作为 item 参数传递给 create_item 函数。
* 我们还定义了一个 ItemUpdate 模型，用于更新操作，其中所有字段都是可选的。
* 在 update_item 函数中，我们同时接收路径参数 item_id 和请求体 item: ItemUpdate。
运行应用，访问 http://127.0.0.1:8000/docs。你会看到 /items/ 和 /items/{item_id} 的 POST/PUT 接口已经出现在文档中。你可以点击 /items/ (POST)，然后点击 “Try it out”，在请求体中输入 JSON 数据进行测试。尝试发送一个缺少 name 或 price 的请求体，看看返回的错误信息。
json
{
"name": "Foo",
"description": "A very nice Item",
"price": 35.4,
"tax": 3.2
}
发送上述 JSON 会成功，返回：
json
{
"name": "Foo",
"description": "A very nice Item",
"price": 35.4,
"tax": 3.2,
"price_with_tax": 38.6
}
发送一个无效的 JSON，例如：
json
{
"description": "Missing name and price"
}
你会收到一个 422 错误响应，包含详细的验证失败信息。
这种通过类型提示结合 Pydantic 模型实现的自动验证和序列化，是 FastAPI 提高开发效率的核心特性之一。它将数据验证的逻辑从业务代码中分离出来，并自动生成文档，极大地减少了重复劳动。
更复杂的 Pydantic 模型和数据验证
Pydantic 提供了丰富的数据类型和验证功能。你可以使用标准的 Python 类型，也可以使用 Pydantic 的特定类型和 Field 函数进行更精细的控制。
“`python
from pydantic import BaseModel, Field
from typing import List, Set, Dict, Optional
class Image(BaseModel):
url: str = Field(…, example=”http://example.com/image.png”) # … 表示字段是必需的
name: str = Field(…, example=”The image name”)
class ItemWithMoreDetails(BaseModel):
name: str = Field(
…,
title=”Item Name”,
description=”The name of the item”,
max_length=50,
example=”Awesome Gadget” # example 会出现在文档中
)
description: Optional[str] = Field(
None,
title=”Item Description”,
max_length=300,
example=”This is a description of the awesome gadget.”
)
price: float = Field(
…,
title=”Item Price”,
description=”The price of the item in USD”,
gt=0, # 大于 0
le=10000, # 小于等于 10000
example=99.99
)
tax: Optional[float] = Field(
None,
gt=0,
le=1000,
example=9.99
)
tags: Set[str] = Field(
set(), # 默认值为空集合
example={“electronics”, “gadget”}
) # 集合类型会自动去重
image: Optional[Image] = None # 嵌套模型
images: Optional[List[Image]] = None # 列表嵌套模型
properties: Dict[str, float] = Field(
{}, # 默认值为空字典
example={“weight_kg”: 0.5, “dimensions_cm”: 10.5}
) # 字典类型
… (在 FastAPI 应用中定义路径操作函数)
@app.post(“/items/complex/”)
async def create_complex_item(item: ItemWithMoreDetails):
return item.model_dump()
“`
Field(...)：用于为 Pydantic 模型字段提供额外的配置，例如默认值、别名、标题、描述、示例、验证约束（如gt,lt,ge,le,min_length,max_length,regex等）。...表示该字段是必需的，没有默认值。example：在Field中提供的示例值会显示在自动生成的文档中。List,Set,Dict: 可以使用 Python 的标准类型提示来定义列表、集合、字典等结构。Optional: 使用typing.Optional或| None来表示字段是可选的。- 嵌套模型: Pydantic 模型可以包含其他 Pydantic 模型，轻松定义复杂的嵌套数据结构。
通过这些强大的特性，你可以用 Pydantic 定义非常灵活和严谨的数据结构，而 FastAPI 会自动利用这些定义来生成文档和处理数据验证。
第三步：异步操作和性能优化
FastAPI 是一个 ASGI 框架，天生支持异步。使用 async def 定义的路径操作函数可以在等待一些慢速操作（如数据库查询、外部 API 调用、文件 I/O）时，释放工作线程去处理其他请求。这对于构建高并发的 API 至关重要，特别是 I/O 密集型应用。
例如，一个模拟异步数据库调用的例子：
“`python
import asyncio
from fastapi import FastAPI
app = FastAPI()
模拟一个异步数据库查询
async def fake_db_query(item_id: int):
await asyncio.sleep(1) # 模拟异步等待，例如等待数据库响应
return {“id”: item_id, “name”: f”Item {item_id}”, “value”: item_id * 10}
@app.get(“/items/{item_id}”)
async def read_item_from_db(item_id: int):
“””
从模拟数据库获取物品信息 (异步操作)
“””
item = await fake_db_query(item_id) # 等待异步操作完成
return item
如果你有一些老的同步代码或者库，它们不是异步的
你仍然可以在 FastAPI 中使用同步函数
import time
def fake_sync_process():
time.sleep(1) # 模拟同步阻塞操作
return “sync result”
@app.get(“/sync_process/”)
def run_sync_process(): # 注意这里是 def 而不是 async def
“””
运行一个同步阻塞操作
“””
# FastAPI 会自动在一个单独的线程池中运行这个同步函数，
# 从而不会阻塞主事件循环。
result = fake_sync_process()
return {“result”: result}
“`
- 在
read_item_from_db函数中，await asyncio.sleep(1)模拟了一个耗时 1 秒的异步操作。当执行到await时，这个函数会暂停执行，但不会阻塞整个进程。FastAPI/Uvicorn 可以继续处理其他传入的请求。1 秒后，当fake_db_query完成时，read_item_from_db会恢复执行。 run_sync_process函数使用了def而不是async def。在 ASGI 应用中直接运行同步阻塞代码会阻塞整个事件循环，导致无法处理其他请求。FastAPI 非常智能地处理了这种情况：当它检测到一个def定义的路径操作函数时，它会自动在一个内部线程池中运行这个函数，从而避免阻塞主事件循环。这意味着你可以在 FastAPI 中无缝地混合使用异步和同步代码，而 FastAPI 会负责安排它们的执行。
核心原则：
* 如果你的函数内部使用了 await 来调用其他异步函数（如 await db.fetch(...) 或 await http_client.get(...)），则必须使用 async def 定义函数。
* 如果你的函数内部执行的是 CPU 密集型计算或者调用了会阻塞进程的同步 I/O 操作（如 time.sleep() 或调用某些同步库），则应该使用 def 定义函数。FastAPI 会负责在线程池中运行它。
* 如果你的函数既不 await 也不执行阻塞的同步 I/O，那么使用 async def 或 def 都可以，性能上差异不大，但推荐 async def 以保持风格一致，并方便未来引入异步操作。
正确地使用 async def 和 await 是构建高性能 FastAPI 应用的关键。它使得应用能够在等待外部资源时保持响应性，从而处理更多的并发请求。
第四步：依赖注入系统
FastAPI 提供了一个非常强大且易于使用的依赖注入（Dependency Injection，简称 DI）系统。依赖注入是一种设计模式，它使得组件（在这里是路径操作函数）可以声明它们所需的依赖（其他函数或对象），而不是自己创建或查找这些依赖。FastAPI 负责“注入”这些依赖。
依赖注入有许多优点：
* 代码复用：可以将一些公共逻辑（如数据库连接、用户认证、权限检查、获取配置）封装成一个“依赖函数”，然后在多个路径操作函数中复用。
* 代码组织：将复杂的逻辑分解成更小的、可管理的依赖。
* 可测试性：测试时可以轻松地用模拟（Mock）依赖替换真实依赖，方便进行单元测试。
* 解耦：路径操作函数不直接与外部资源耦合，降低了组件之间的依赖。
在 FastAPI 中，你通过在路径操作函数的参数中使用 Depends 来声明依赖。Depends 接收一个可调用的对象（通常是一个函数）。
“`python
from fastapi import FastAPI, Depends, Header, HTTPException, status
from typing import Annotated # Python 3.9+ 支持 typing.Annotated
app = FastAPI()
这是一个简单的依赖函数，用于验证一个 token
async def verify_token(token: str = Header(…)):
“””
从请求头中提取并验证 token
“””
# 模拟 token 验证逻辑
if token != “fake-super-secret-token”:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Invalid token”,
headers={“WWW-Authenticate”: “Bearer”},
)
# 如果验证通过，依赖函数可以返回一个值
return token # 或者返回一个用户对象等
这是一个更复杂的依赖，使用 yield 来处理资源的创建和清理
async def get_db_connection():
“””
获取数据库连接 (模拟)
“””
print(“Connecting to DB…”)
db = {“data”: “some data”} # 模拟数据库连接对象
try:
yield db # 在 yield 之前执行 setup 逻辑
finally:
# yield 之后执行 cleanup 逻辑，即使发生异常也会执行
print(“Closing DB connection…”)
# db.close() # 真实的数据库连接清理
使用 Annotated 结合 Depends (更现代的方式，推荐 Python 3.9+)
@app.get(“/items/”)
async def read_items(
# 声明依赖：token 需要通过 verify_token 函数获取
# 如果 verify_token 抛出 HTTPException，请求会中止
token: Annotated[str, Depends(verify_token)],
# 声明依赖：db 需要通过 get_db_connection 函数获取
# get_db_connection 会在请求处理期间提供连接
db: Annotated[dict, Depends(get_db_connection)]
):
“””
获取物品列表，需要 token 认证和数据库连接
“””
# 在这里可以直接使用 token 和 db 对象
print(f”Using token: {token}”)
print(f”Using db: {db}”)
return {“items”: [db[“data”]]}
使用旧的 Depends 语法 (兼容性更好，Python 3.8 及以下)
@app.get(“/items_old/”)
async def read_items_old(
token: str = Depends(verify_token),
db: dict = Depends(get_db_connection)
):
“””
旧语法获取物品列表
“””
print(f”Using token: {token}”)
print(f”Using db: {db}”)
return {“items”: [db[“data”]]}
不使用认证的公开接口
@app.get(“/public/”)
async def public_endpoint():
return {“message”: “This is a public endpoint”}
“`
在这个例子中：
* verify_token 函数是一个依赖。它从请求头中获取 token，进行验证，如果失败则抛出 HTTPException。如果成功，它返回 token 字符串。
* get_db_connection 函数是一个使用 yield 的依赖。它模拟了数据库连接的建立 (print("Connecting to DB..."))，通过 yield db 将连接对象 db 提供给依赖它的路径操作函数。在路径操作函数执行完毕（无论成功还是失败），finally 块中的清理逻辑 (print("Closing DB connection...")) 会被执行。这非常适合管理有生命周期的资源，如数据库会话、文件句柄等。
* 在 read_items 路径操作函数中，token: Annotated[str, Depends(verify_token)] 和 db: Annotated[dict, Depends(get_db_connection)] 声明了该函数依赖于 verify_token 和 get_db_connection。
* 当收到对 /items/ 的请求时，FastAPI 会先调用 verify_token 和 get_db_connection。
* 如果 verify_token 抛出异常，请求会在那里中止，不会调用 read_items。
* 如果所有依赖都成功执行，它们的返回值（或 yield 的值）会作为参数传递给 read_items 函数。
* verify_token 函数中的 token: str = Header(...) 也是一个依赖的例子——路径操作函数依赖于从请求头中获取 token。FastAPI 内置了许多这样的依赖（如 Header, Cookie, Body, Query, Path, File, Form）。
Annotated vs 旧语法 param: Type = Depends(...): Annotated (Python 3.9+) 提供了一种更清晰的方式来为类型添加元数据，而不会影响类型本身。在 FastAPI 中，它被用来标记一个参数是通过 Depends 提供的依赖。这是官方推荐的现代写法，但旧的 = Depends(...) 语法在所有 Python 3.7+ 版本中都可用。
依赖注入系统是 FastAPI 的核心设计之一，它使得构建模块化、可测试和易于维护的应用变得非常容易。你可以将复杂的业务逻辑、基础设施交互封装在依赖中，然后在需要的地方轻松地“注入”它们。
第五步：更高级的特性和应用组织
错误处理
FastAPI 自动处理 Pydantic 验证错误、路径参数/查询参数转换错误等，并返回 422 状态码。对于其他 HTTP 异常，可以使用 fastapi.HTTPException 手动抛出：
“`python
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
items_db = {“foo”: “The Foo Fighters”}
@app.get(“/items/{item_id}”)
async def read_item(item_id: str):
if item_id not in items_db:
# 抛出 HTTPException，FastAPI 会将其转换为相应的 HTTP 响应
raise HTTPException(status_code=404, detail=”Item not found”)
return {“item”: items_db[item_id]}
也可以为特定的 HTTP 状态码定义响应模型和描述
@app.get(
“/items/{item_id}/detailed”,
response_model=dict, # 成功的响应模型
responses={
404: {“description”: “Item not found”} # 额外定义的响应
}
)
async def read_item_detailed(item_id: str):
if item_id not in items_db:
raise HTTPException(status_code=404, detail=”Item not found”)
return {“item”: items_db[item_id]}
自定义异常处理
from fastapi import Request, Response
from fastapi.exception_handlers import http_exception_handler, request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
# 这里可以自定义处理 Pydantic 验证错误 (422) 的方式
print(f”OMG! The client sent invalid data: {exc.errors()}”)
return await request_validation_exception_handler(request, exc) # 调用默认处理函数
也可以为自定义异常定义处理函数
class MyCustomException(Exception):
def init(self, name: str):
self.name = name
@app.exception_handler(MyCustomException)
async def custom_exception_handler(request: Request, exc: MyCustomException):
return Response(content=f”Oops! My custom error: {exc.name}”, status_code=418) # 418 I’m a teapot
@app.get(“/teapot_error/”)
async def trigger_custom_error():
raise MyCustomException(“This is a teapot error”)
“`
raise HTTPException(...)是在路径操作函数中返回错误响应的标准方式。你可以指定状态码 (status_code)、详细信息 (detail)，还可以添加自定义头部 (headers)。- 通过
@app.exception_handler(ExceptionType)装饰器，可以注册自定义的异常处理函数，用于捕获特定类型的异常并返回自定义的响应。
组织大型应用：APIRouter
随着应用增长，所有路径操作函数都放在一个文件中的 app = FastAPI() 实例下会变得难以管理。FastAPI 提供了 APIRouter 来帮助组织代码。
你可以将相关的路径操作函数分组到不同的 APIRouter 实例中，然后在主 FastAPI 应用中包含这些路由器。
创建一个 routers 目录，并在其中创建文件，例如 routers/items.py 和 routers/users.py。
routers/items.py:
“`python
routers/items.py
from fastapi import APIRouter, HTTPException, Path, Query, status
from pydantic import BaseModel
from typing import Optional
创建一个 APIRouter 实例
router = APIRouter(
prefix=”/items”, # 为所有路径操作添加统一的前缀
tags=[“items”], # 为这个路由器下的所有接口添加标签，用于文档分组
dependencies=[], # 可以为这个路由器下的所有接口添加共同的依赖
responses={404: {“description”: “Item not found in this router”}} # 可以定义路由器级别的响应
)
items_db = {“foo”: “The Foo Fighters”}
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None
@router.get(“/”)
async def read_items():
return list(items_db.keys())
@router.get(“/{item_id}”)
async def read_item(
item_id: str = Path(…, title=”The ID of the item to get”), # Path 可以提供更多元数据
q: Optional[str] = Query(None, alias=”item-query”, title=”Query string for the item”) # Query 可以设置别名、元数据
):
if item_id not in items_db:
raise HTTPException(status_code=404, detail=”Item not found”)
return {“item”: items_db[item_id]}
@router.post(“/”)
async def create_item(item: Item):
# 简单的模拟创建
items_db[item.name.lower().replace(” “, “-“)] = item.description
return item
“`
routers/users.py:
“`python
routers/users.py
from fastapi import APIRouter
router = APIRouter(
prefix=”/users”,
tags=[“users”],
)
@router.get(“/”)
async def read_users():
return [“Rick”, “Morty”]
@router.get(“/{user_id}”)
async def read_user(user_id: int):
return {“user_id”: user_id}
“`
main.py:
“`python
main.py
from fastapi import FastAPI
from routers import items, users # 导入路由器模块
app = FastAPI(
title=”My Awesome API”, # 为整个 API 添加标题和描述
description=”This is a very awesome API with items and users.”,
version=”0.0.1″,
# OpenAPI 文档相关的额外参数
openapi_url=”/api/v1/openapi.json”, # 可以修改 OpenAPI JSON 的路径
docs_url=”/documentation”, # 可以修改 Swagger UI 文档的路径
redoc_url=None, # 可以禁用 ReDoc 文档
)
包含路由器
app.include_router(items.router)
app.include_router(users.router)
根路径仍然保留在 main.py 中
@app.get(“/”)
async def read_root():
return {“message”: “Welcome to My Awesome API!”}
“`
运行 uvicorn main:app --reload。访问 /docs，你会看到 API 文档现在按照 “items” 和 “users” 标签分组了。访问 /items/foo 或 /users/1 验证路由是否生效。
通过 APIRouter，你可以：
* 为一组相关的接口设置统一的 prefix，避免重复写路径前缀。
* 使用 tags 为文档分组，提高文档的可读性。
* 为路由器下的所有接口应用共同的 dependencies。
* 定义路由器级别的响应（responses）。
* 将应用拆分成多个文件和模块，提高代码的可维护性。
中间件 (Middleware)
中间件是在请求被路径操作函数处理之前以及路径操作函数返回响应之后执行的函数。它们用于实现一些横切关注点，例如：
* CORS (跨域资源共享)
* GZip 压缩
* 认证/授权检查 (除了依赖注入之外的方式)
* 日志记录
* 修改请求或响应头部
FastAPI 基于 Starlette，可以直接使用 Starlette 的中间件。
“`python
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
其他可能的中间件：
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time
app = FastAPI()
添加 CORS 中间件
允许来自所有源（*）的跨域请求，支持所有方法和头部
app.add_middleware(
CORSMiddleware,
allow_origins=[““], # 或者指定允许的源列表，例如 [“http://localhost:3000”, “https://example.com”]
allow_credentials=True, # 允许携带 cookie
allow_methods=[““], # 允许所有 HTTP 方法，或者指定列表 [“GET”, “POST”]
allow_headers=[“*”], # 允许所有头部，或者指定列表 [“Content-Type”, “Authorization”]
)
自定义中间件示例 (记录请求处理时间)
@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
@app.get(“/”)
async def read_root():
await asyncio.sleep(0.1) # 模拟一些处理时间
return {“message”: “Hello, World!”}
“`
- 使用
app.add_middleware()添加中间件。中间件是按照添加的顺序执行的（请求进入时），响应返回时按照相反的顺序执行。 CORSMiddleware是 FastAPI/Starlette 提供的一个常用中间件，用于处理跨域请求。- 自定义中间件是一个
async def函数，接收request: Request和call_next参数。call_next是下一个中间件或路径操作函数。await call_next(request)是调用链中的下一个组件。
测试
FastAPI 应用非常容易测试。你可以使用 fastapi.testclient.TestClient 来发送请求并检查响应。TestClient 使用 httpx 库（需要安装 pip install httpx）在内存中模拟 HTTP 请求，无需运行实际的服务器。
“`python
test_main.py
from fastapi.testclient import TestClient
导入你的 FastAPI 应用实例
from main import app
创建 TestClient 实例
client = TestClient(app)
def test_read_root():
# 发送 GET 请求到 / 路径
response = client.get(“/”)
# 检查响应状态码
assert response.status_code == 200
# 检查响应体内容
assert response.json() == {“message”: “Welcome to My Awesome API!”} # 根据你的 main.py 修改
def test_read_item():
response = client.get(“/items/foo”)
assert response.status_code == 200
assert response.json() == {“item”: “The Foo Fighters”} # 根据你的 routers/items.py 中的 items_db 修改
def test_read_nonexistent_item():
response = client.get(“/items/baz”)
assert response.status_code == 404
assert response.json() == {“detail”: “Item not found”} # 根据 routers/items.py 中的 HTTPException 定义修改
def test_create_item():
response = client.post(
“/items/”,
json={“name”: “Bar”, “description”: “A Bar item”, “price”: 10.99},
)
assert response.status_code == 200
# 检查响应体是否包含创建的物品信息 (根据 routers/items.py 中的 create_item 返回值修改)
assert response.json() == {
“name”: “Bar”,
“description”: “A Bar item”,
“price”: 10.99,
“tax”: None
}
def test_create_item_invalid():
response = client.post(
“/items/”,
json={“description”: “Missing name and price”}, # 无效请求体
)
assert response.status_code == 422 # 验证错误状态码
测试需要认证的接口 (假设需要 Header 中包含 “token”: “fake-super-secret-token”)
def test_read_items_authenticated():
response = client.get(
“/items/”, # 假设这个接口在 routers/items.py 中，并且需要 token 依赖
headers={“token”: “fake-super-secret-token”} # 提供正确的 token
)
assert response.status_code == 200
# assert response.json() == … # 验证返回的数据
def test_read_items_unauthenticated():
response = client.get(
“/items/”,
headers={“token”: “wrong-token”} # 提供错误的 token
)
assert response.status_code == 401 # 认证失败状态码
assert response.json() == {“detail”: “Invalid token”} # 验证错误信息
“`
运行 pytest 命令来执行这些测试（需要安装 pytest：pip install pytest httpx）。FastAPI 的测试客户端使得编写集成测试变得非常简单直观。
部署
FastAPI 应用是 ASGI 应用，需要使用 ASGI 服务器来运行，例如 Uvicorn、Hypercorn。
对于简单的部署，可以直接使用 Uvicorn：
bash
uvicorn main:app
对于生产环境，通常需要更 robust 的方案，例如使用 Gunicorn 作为进程管理器，再由 Gunicorn 启动多个 Uvicorn worker：
“`bash
安装 gunicorn
pip install gunicorn
使用 gunicorn + uvicorn workers 启动
-w 4 表示启动 4 个 worker 进程
-k uvicorn 表示使用 uvicorn worker class
gunicorn main:app -w 4 -k uvicorn
“`
更现代的部署方式通常涉及 Docker。你可以创建一个 Dockerfile 来打包你的应用和依赖，然后在容器中运行。
“`dockerfile
Dockerfile
使用一个轻量级的 Python 镜像
FROM python:3.10-slim
设置工作目录
WORKDIR /code
复制依赖文件并安装依赖
COPY requirements.txt /code/
RUN pip install –no-cache-dir -r requirements.txt
复制应用代码
COPY . /code/
暴露应用端口 (根据你的需要修改)
EXPOSE 80
启动应用 (生产环境建议使用 gunicorn + uvicorn)
CMD [“gunicorn”, “main:app”, “–workers”, “4”, “–worker-class”, “uvicorn.workers.UvicornWorker”, “–bind”, “0.0.0.0:80”]
如果是简单的测试，也可以直接用 uvicorn
CMD [“uvicorn”, “main:app”, “–host”, “0.0.0.0”, “–port”, “80”]
“`
requirements.txt 应该包含你的项目依赖，例如：
fastapi
uvicorn[standard]
pydantic
httpx # 如果需要测试
构建并运行 Docker 镜像：
bash
docker build -t my-fastapi-app .
docker run -p 80:80 my-fastapi-app
这只是一个简单的概述，实际生产部署可能还需要考虑 SSL/TLS 终止、负载均衡、容器编排（Kubernetes, Docker Swarm）等。
其他特性
FastAPI 还支持许多其他有用的特性，例如：
* 状态管理：使用依赖注入可以方便地管理应用状态或共享对象。
* 后台任务 (Background Tasks)：允许你在返回 HTTP 响应后执行一些不需要客户端等待的任务。
* WebSocket：支持 WebSocket 通信。
* GraphQL：可以与其他库集成支持 GraphQL。
* 子路径挂载：可以将其他 ASGI 应用（如 Flask, Django 应用，或其他 FastAPI 应用）挂载到子路径下。
结论：快速、高效、现代的 API 开发利器
FastAPI 凭借其卓越的性能、极高的开发效率、自动化的文档和数据验证、强大的依赖注入系统以及对异步编程的良好支持，迅速成为 Python API 开发领域的一颗璀璨新星。
通过本文的介绍，你应该已经对 FastAPI 的核心概念和主要特性有了深入的理解：
* 它如何利用类型提示和 Pydantic 实现自动数据验证和序列化。
* 它如何自动生成交互式 API 文档，极大地提升开发和协作效率。
* 它如何通过异步支持和智能的同步代码处理，实现高性能。
* 它如何通过强大的依赖注入系统，让代码更具模块化、可测试性和可复用性。
* 它如何通过 APIRouter 组织大型应用，以及如何进行错误处理、中间件配置和测试。
FastAPI 不仅仅是一个 Web 框架，它是一整套提高 API 开发效率和质量的解决方案。无论你是要构建一个简单的微服务，还是一个复杂的企业级 API 网关，FastAPI 都能成为你的得力助手。
如果你正在寻找一个现代、高性能、易于学习且功能强大的 Python Web 框架来构建 API，那么 FastAPI 绝对值得你投入时间去学习和使用。开始你的 FastAPI 之旅吧，你会发现 API 开发从未如此高效和愉快！