使用 Python FastAPI 构建 RESTful API:新手指南
使用 Python FastAPI 构建 RESTful API:新手终极指南
在现代 Web 开发中,API(应用程序编程接口)扮演着至关重要的角色。它们是不同软件系统之间通信的桥梁,使得前端应用(如网站、移动 App)能够与后端服务交互、获取数据或执行操作。REST (Representational State Transfer) 是一种流行的架构风格,用于设计网络应用程序,而 RESTful API 则是遵循 REST 原则构建的 API。
Python 作为一种功能强大且易于学习的语言,拥有众多优秀的 Web 框架。近年来,FastAPI 凭借其卓越的性能、易用性以及现代化的特性(如异步支持和自动文档生成),迅速成为构建 API 的热门选择。本指南将带你从零开始,一步步学习如何使用 Python 和 FastAPI 构建高效、健壮的 RESTful API。
一、 为什么选择 FastAPI?
在深入实践之前,让我们先了解一下 FastAPI 脱颖而出的原因:
- 极高的性能: FastAPI 基于 Starlette (ASGI 框架) 和 Pydantic (数据验证库),其性能与 NodeJS 和 Go 不相上下,是现有 Python 框架中速度最快的之一。这得益于其底层的异步特性和优化的设计。
- 快速开发: FastAPI 旨在提高开发效率。其直观的语法、强大的类型提示集成以及自动生成的文档,可以让你将编码时间缩短高达 200%-300%。
- 更少的 Bug: 利用 Python 的类型提示,FastAPI 在运行时进行数据验证,并在开发过程中提供出色的编辑器支持(如自动补全、类型检查)。这大大减少了由于数据类型错误或缺失导致的 Bug。
- 易于学习和使用: FastAPI 的设计简洁明了,学习曲线相对平缓。官方文档非常详尽,并提供了大量示例。
- 自动交互式文档: 无需额外配置,FastAPI 会自动为你生成基于 OpenAPI (前身为 Swagger) 和 JSON Schema 标准的交互式 API 文档 (Swagger UI 和 ReDoc)。这对于 API 的测试、调试和团队协作非常有价值。
- 基于标准: 完全兼容 OpenAPI 和 JSON Schema,这意味着你的 API 可以轻松地与各种工具和服务集成。
- 异步支持: 内建对 Python
async/await
语法的完美支持,非常适合构建需要处理高并发 I/O 密集型任务(如数据库查询、外部 API 调用)的应用程序。 - 依赖注入系统: 提供了一个简单而强大的依赖注入系统,有助于编写可重用、可测试和易于管理的代码。
二、 准备工作:环境设置
在开始编码之前,你需要确保你的开发环境已经准备就绪。
-
安装 Python: FastAPI 需要 Python 3.7+ 版本。如果你的系统尚未安装或版本过低,请前往 Python 官网 (python.org) 下载并安装最新稳定版本。你可以通过在终端运行
python --version
或python3 --version
来检查你的 Python 版本。 -
创建虚拟环境 (强烈推荐): 为了保持项目依赖的隔离性,避免不同项目间的库版本冲突,强烈建议为每个项目创建一个独立的虚拟环境。
- 打开你的终端或命令行工具。
- 导航到你希望创建项目的目录。
- 运行以下命令创建虚拟环境(这里我们命名为
venv
):
bash
python -m venv venv
# 或者 python3 -m venv venv - 激活虚拟环境:
- Windows:
bash
.\venv\Scripts\activate - macOS / Linux:
bash
source venv/bin/activate
激活成功后,你会在命令行提示符前看到(venv)
字样。
- Windows:
-
安装 FastAPI 和 Uvicorn:
- 确保你的虚拟环境已激活。
- 使用 pip (Python 的包管理器) 安装 FastAPI 和 Uvicorn。Uvicorn 是一个 ASGI 服务器,用于运行你的 FastAPI 应用。
bash
pip install fastapi uvicorn[standard]
uvicorn[standard]
会安装 Uvicorn 以及一些推荐的依赖(如uvloop
和httptools
),以获得更好的性能。
现在,你的开发环境已经准备好了!
三、 你的第一个 FastAPI 应用
让我们从一个最简单的 "Hello World" 示例开始。
-
创建
main.py
文件: 在你的项目目录下创建一个名为main.py
的 Python 文件。 -
编写代码: 在
main.py
文件中输入以下代码:```python
from fastapi import FastAPI1. 创建 FastAPI 实例
app
将是我们 API 的主要交互点app = FastAPI()
2. 定义路径操作 (Path Operation)
@app.get("/")
是一个 "装饰器 (decorator)"它告诉 FastAPI 下面的函数
read_root
负责处理对路径 "/" 的 GET 请求@app.get("/")
async def read_root():
# 3. 路径操作函数 (Path Operation Function)
# 这是一个标准的 Python 异步函数 (async def)
# FastAPI 也支持普通的同步函数 (def)
# 函数返回的内容将作为 API 响应发送给客户端
# FastAPI 会自动将 Python 字典或列表转换为 JSON 格式
return {"message": "Hello World"}4. 定义另一个路径操作
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None):
# 这里的{item_id}
是一个 "路径参数 (Path Parameter)"
#item_id: int
使用 Python 类型提示,FastAPI 会自动进行数据转换和验证
#q: str | None = None
是一个 "查询参数 (Query Parameter)"
#str | None
表示 q 可以是字符串,也可以是 None (可选)
#= None
为其设置了默认值
response = {"item_id": item_id}
if q:
response.update({"q": q})
return response
``` -
运行应用:
- 回到你的终端 (确保虚拟环境已激活并且你在
main.py
文件所在的目录)。 -
运行 Uvicorn 服务器:
bash
uvicorn main:app --reloadmain
: 指的是main.py
文件 (Python 模块)。app
: 指的是你在main.py
文件中创建的FastAPI()
实例对象。--reload
: 这个参数非常有用,它会让服务器在代码更改后自动重启,方便开发。
-
运行成功后,你会看到类似以下的输出:
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [xxxxx] using statreload
INFO: Started server process [xxxxx]
INFO: Waiting for application startup.
INFO: Application startup complete.
这表示你的 API 服务器正在本地http://127.0.0.1:8000
上运行。
- 回到你的终端 (确保虚拟环境已激活并且你在
-
测试 API:
- 根路径 (
/
): 打开你的 Web 浏览器,访问http://127.0.0.1:8000
。你应该会看到 JSON 响应:{"message": "Hello World"}
。 - 带路径参数 (
/items/{item_id}
): 访问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"}
。 - 尝试无效路径参数: 访问
http://127.0.0.1:8000/items/foo
。你会看到一个由 FastAPI 自动生成的清晰错误信息,因为 "foo" 不是一个有效的整数。
- 根路径 (
恭喜!你已经成功创建并运行了你的第一个 FastAPI 应用!
四、 深入理解核心概念
上面的例子已经展示了 FastAPI 的一些核心概念,让我们更详细地了解它们。
-
路径操作 (Path Operations):
- 路径 (Path): 指 URL 中域名之后的部分,如
/
或/items/{item_id}
。 - 操作 (Operation): 指 HTTP 方法,如
GET
,POST
,PUT
,DELETE
,OPTIONS
,HEAD
,PATCH
,TRACE
。 - 路径操作装饰器: FastAPI 使用装饰器将 HTTP 方法和路径绑定到处理函数上,例如
@app.get()
,@app.post()
,@app.put()
,@app.delete()
等。
- 路径 (Path): 指 URL 中域名之后的部分,如
-
路径参数 (Path Parameters):
- 用于捕获 URL 路径中特定段的值。
- 使用花括号
{}
在路径字符串中定义,如/users/{user_id}
。 - 路径参数的值会作为参数传递给路径操作函数。
- 结合 Python 类型提示 (如
user_id: int
),FastAPI 会自动进行数据验证和转换。如果传入的值无法转换为指定类型(例如,给需要int
的地方传入了字符串 "abc"),FastAPI 会返回一个带有清晰错误信息的 422 Unprocessable Entity 响应。
-
查询参数 (Query Parameters):
- URL 中
?
之后的部分,用于传递额外的键值对信息,如/items?skip=0&limit=10
。 - 在路径操作函数中声明不属于路径参数的函数参数,FastAPI 会自动将其识别为查询参数。
- 可以使用类型提示进行验证 (
limit: int
)。 - 可以设置默认值 (
skip: int = 0
),使其成为可选参数。 - 对于可选参数,可以使用
| None
(Python 3.10+) 或Optional[type]
(较早版本) 来显式声明,并设置默认值为None
(q: str | None = None
)。
- URL 中
-
请求体 (Request Body):
- 当客户端需要向 API 发送数据时(通常在
POST
,PUT
,PATCH
请求中),数据会包含在请求体中,通常是 JSON 格式。 - FastAPI 使用 Pydantic 库来定义、验证和处理请求体数据。
- 当客户端需要向 API 发送数据时(通常在
五、 使用 Pydantic 进行数据建模与验证
Pydantic 是 FastAPI 的核心依赖之一,它是一个基于 Python 类型提示的数据验证和设置管理库。
-
定义 Pydantic 模型:
- 创建一个继承自
pydantic.BaseModel
的类。 - 在类中定义属性,并使用 Python 类型提示来指定它们的类型。
```python
from pydantic import BaseModel, Field
from typing import List, Optionalclass Item(BaseModel):
name: str
description: str | None = None # 可选字段,默认 None
price: float
tax: float | None = None
tags: List[str] = [] # 列表字段,默认为空列表更复杂的例子,使用 Field 进行额外验证
class User(BaseModel):
username: str
email: str
full_name: str | None = None
age: int = Field(default=..., gt=0, le=120) # default=... 表示必填, gt=0 大于0, le=120 小于等于120
``` - 创建一个继承自
-
在路径操作中使用模型:
- 要接收请求体,只需在路径操作函数中声明一个参数,并将其类型提示为你的 Pydantic 模型。
```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModelapp = FastAPI()
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None模拟数据库存储
fake_items_db = {}
@app.post("/items/", status_code=201) # 指定成功响应的状态码
async def create_item(item: Item):
# FastAPI 会自动:
# 1. 从请求体中读取 JSON 数据。
# 2. 使用 Item 模型验证数据类型和约束。
# 3. 如果数据无效,返回 422 错误。
# 4. 如果数据有效,将 JSON 数据转换为 Item 对象,并传递给item
参数。
item_dict = item.dict()
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({"price_with_tax": price_with_tax})# 简单的存储逻辑 (实际应用会用数据库) item_id = len(fake_items_db) + 1 fake_items_db[item_id] = item_dict return {"item_id": item_id, **item_dict} # 返回创建的 Item 数据及 ID
@app.get("/items/{item_id}")
async def get_item(item_id: int):
if item_id not in fake_items_db:
# 主动抛出 HTTP 异常
raise HTTPException(status_code=404, detail="Item not found")
return fake_items_db[item_id]
```- 数据验证: 如果客户端发送的 JSON 数据不符合
Item
模型的定义(例如,缺少name
或price
字段,或者price
不是数字),FastAPI 会自动返回一个详细的 422 Unprocessable Entity 错误响应,指出哪些字段有问题。 - 编辑器支持: 在你的代码编辑器中,当你访问
item.
时,你会得到所有模型属性(name
,description
,price
,tax
)的自动补全和类型检查。
六、 自动交互式 API 文档
这是 FastAPI 最令人惊叹的功能之一。在你运行应用后:
-
Swagger UI: 访问
http://127.0.0.1:8000/docs
。你会看到一个交互式的 API 文档界面。- 它列出了你所有的 API 路径和操作。
- 你可以展开每个操作查看其详细信息,包括:
- 描述 (如果通过
description
参数添加到装饰器或函数文档字符串中)。 - 路径参数、查询参数及其类型和是否必需。
- 请求体模型 (如果有的话),包括每个字段的类型、约束和示例值。
- 可能的响应状态码和响应模型。
- 描述 (如果通过
- 最棒的是,你可以直接在这个界面上 测试你的 API!输入参数,点击 "Execute",它会向你的 API 发送请求并显示响应。
-
ReDoc: 访问
http://127.0.0.1:8000/redoc
。你会看到另一种风格的、更侧重于文档展示的界面。
这一切都是自动生成的,基于你的代码、类型提示和 Pydantic 模型。这极大地简化了 API 的文档编写和维护工作,并为 API 的使用者(包括前端开发者或第三方集成者)提供了极大的便利。
七、 响应模型 (Response Model)
有时,你希望 API 返回的数据结构与内部处理或存储的数据结构有所不同,或者你只想返回 Pydantic 模型中的部分字段,或者确保返回的数据符合特定模式。这时可以使用 response_model
参数。
```python
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
app = FastAPI()
class UserBase(BaseModel):
username: str
email: EmailStr # Pydantic 提供的特殊 Email 类型
full_name: str | None = None
class UserIn(UserBase): # 用于输入的模型
password: str
class UserOut(UserBase): # 用于输出的模型 (不含密码)
pass # 继承 UserBase 的所有字段
假设这是数据库模型,可能包含更多字段
class UserInDB(UserBase):
hashed_password: str
模拟数据库
fake_user_db = {
"johndoe": {
"username": "johndoe",
"email": "[email protected]",
"full_name": "John Doe",
"hashed_password": "fakehashedpassword" # 不应返回给客户端
}
}
@app.post("/users/", response_model=UserOut, status_code=201)
async def create_user(user: UserIn):
# 实际应用中会 hash 密码并存储
user_in_db = UserInDB(**user.dict(), hashed_password=f"hashed_{user.password}")
fake_user_db[user.username] = user_in_db.dict()
# 即使内部返回了包含 hashed_password 的字典,
# FastAPI 也会根据 response_model=UserOut 进行过滤
return user_in_db
@app.get("/users/{username}", response_model=UserOut)
async def read_user(username: str):
if username not in fake_user_db:
raise HTTPException(status_code=404, detail="User not found")
user_data = fake_user_db[username]
# 返回完整数据,FastAPI 会过滤
return user_data
```
在上面的例子中:
* create_user
和 read_user
都指定了 response_model=UserOut
。
* 这意味着无论函数实际返回什么样的数据(只要它包含 UserOut
所需的字段),FastAPI 都会确保最终发送给客户端的 JSON 响应只包含 UserOut
模型中定义的字段 (username
, email
, full_name
),并且数据类型符合模型定义。
* 这有助于:
* 数据过滤: 避免意外泄露敏感信息(如密码哈希)。
* 保证输出结构: 确保 API 响应格式的一致性。
* 自动文档: response_model
也会被包含在自动生成的文档中,清晰地告诉使用者 API 会返回什么样的数据。
八、 错误处理
FastAPI 提供了处理错误的机制:
- 自动验证错误: 如前所述,当路径参数、查询参数或请求体数据验证失败时,FastAPI 会自动返回 422 Unprocessable Entity 错误,并在响应体中包含详细的错误信息。
-
HTTPException
: 当你需要根据业务逻辑主动返回 HTTP 错误时(例如,资源未找到、权限不足),可以导入并raise
HTTPException
。```python
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}@app.get("/secure-data")
async def read_secure_data(api_key: str | None = None):
if api_key != "secret_key": # 简单的权限检查示例
raise HTTPException(status_code=403, detail="Invalid API Key or Forbidden")
return {"data": "Super secret data"}
``
HTTPException接受
status_code(标准的 HTTP 状态码) 和
detail` (错误信息,可以是字符串或 JSON 兼容的对象) 作为参数。FastAPI 会捕获这个异常并生成相应的 HTTP 错误响应。
九、 异步 (async
/await
)
FastAPI 从底层就支持异步操作。如果你的路径操作函数需要执行 I/O 密集型任务(如访问数据库、调用外部 API、读写文件),使用 async def
定义函数并配合 await
调用异步库会带来显著的性能优势。
```python
import asyncio
from fastapi import FastAPI
app = FastAPI()
模拟一个耗时的 I/O 操作 (如数据库查询或外部 API 调用)
async def fetch_data_from_external_source(source_id: int):
print(f"Starting to fetch data for {source_id}...")
await asyncio.sleep(2) # 模拟网络延迟或数据库查询耗时
print(f"Finished fetching data for {source_id}.")
return {"source_id": source_id, "data": f"Some data from source {source_id}"}
@app.get("/data/{source_id}")
async def get_data(source_id: int):
# 使用 await 调用异步函数
result = await fetch_data_from_external_source(source_id)
return result
@app.get("/multiple-data")
async def get_multiple_data():
# 使用 asyncio.gather 并发执行多个异步任务
task1 = fetch_data_from_external_source(1)
task2 = fetch_data_from_external_source(2)
results = await asyncio.gather(task1, task2)
# 即使每个任务耗时 2 秒,由于并发执行,总耗时大约也是 2 秒,而不是 4 秒
return results
```
- 当你使用
async def
定义路径操作函数时,FastAPI 会以异步方式运行它。 - 在函数内部,你可以
await
其他async
函数或支持await
的库(如asyncio
,httpx
,databases
,asyncpg
等)。 - 当遇到
await
时,如果操作尚未完成(例如,等待网络响应),当前函数会暂停执行,事件循环可以去处理其他请求或任务,从而实现高并发。 - 如果你不需要执行耗时的 I/O 操作,或者使用的库本身不支持异步,你仍然可以使用普通的
def
定义路径操作函数,FastAPI 会在一个外部线程池中运行它,避免阻塞事件循环。
十、 项目结构与 APIRouter
当应用变得复杂时,将所有路径操作都放在一个 main.py
文件中会变得难以管理。FastAPI 提供了 APIRouter
来帮助组织代码。
-
创建路由文件: 例如,创建一个
routers
目录,并在其中为不同的功能模块创建 Python 文件(如items.py
,users.py
)。your_project/
├── main.py
├── routers/
│ ├── __init__.py
│ ├── items.py
│ └── users.py
└── venv/ -
在路由文件中定义
APIRouter
:-
routers/items.py
:
```python
from fastapi import APIRouter, HTTPException
from pydantic import BaseModelrouter = APIRouter(
prefix="/items", # 为该路由下的所有路径添加前缀
tags=["items"], # 在 OpenAPI 文档中进行分组
responses={404: {"description": "Not found"}}, # 为所有路径定义通用响应
)class Item(BaseModel):
name: str
price: floatfake_items_db = {"plumbus": {"name": "Plumbus", "price": 3.5}}
@router.get("/{item_id}")
async def read_item(item_id: str):
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail="Item not found")
return fake_items_db[item_id]@router.post("/", response_model=Item, status_code=201)
async def create_item(item: Item):
if item.name in fake_items_db:
raise HTTPException(status_code=400, detail="Item already exists")
fake_items_db[item.name] = item.dict()
return item
``` -
类似地创建
routers/users.py
。
-
-
在主应用 (
main.py
) 中包含路由:-
main.py
:
```python
from fastapi import FastAPI
from .routers import items, users # 使用相对导入app = FastAPI()
包含 items 路由
app.include_router(items.router)
包含 users 路由 (假设已创建)
app.include_router(users.router)
@app.get("/")
async def root():
return {"message": "Welcome to the main API"}
```
-
通过这种方式,你可以将不同功能的 API 逻辑分散到不同的模块中,使得代码更清晰、更易于维护和扩展。
十一、 依赖注入 (Dependency Injection)
FastAPI 拥有一个强大的依赖注入系统,通过 Depends
函数实现。依赖项可以是一个函数、类或其他可调用对象,它们可以在路径操作函数运行之前执行一些逻辑,并将结果(或自身)注入到路径操作函数中。
常见用途包括:
* 数据库连接管理
* 身份验证和授权 (例如,获取当前用户)
* 共享配置或资源
* 可重用的参数验证逻辑
```python
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Annotated # 使用 Annotated 提高可读性 (Python 3.9+)
app = FastAPI()
--- 简单的共享依赖示例 ---
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
定义一个类型别名,使依赖注入更清晰
CommonsDep = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonsDep):
# commons
将包含 common_parameters 函数返回的字典
return {"message": "Reading items", "params": commons}
@app.get("/users/")
async def read_users(commons: CommonsDep):
# 同样可以使用这个共享依赖
return {"message": "Reading users", "params": commons}
--- 模拟身份验证依赖 ---
async def get_current_user(token: str | None = None):
if token != "fake-super-secret-token":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
# 实际应用中会解码 token,查询数据库等
return {"username": "fakeuser", "email": "[email protected]"}
CurrentUserDep = Annotated[dict, Depends(get_current_user)]
@app.get("/users/me")
async def read_users_me(current_user: CurrentUserDep):
# 如果 token 无效,get_current_user 会抛出异常,此函数不会执行
# 如果 token 有效,current_user 将包含 get_current_user 返回的用户信息
return current_user
```
依赖注入是 FastAPI 中一个更高级但非常有用的特性,它能极大地提升代码的可重用性和可测试性。
十二、 下一步:继续探索
本指南涵盖了使用 FastAPI 构建 RESTful API 的基础知识。当你掌握了这些核心概念后,可以进一步探索以下主题:
- 数据库集成: 使用 ORM (如 SQLAlchemy with
databases
orSQLModel
) 或 ODM (如Motor
for MongoDB) 与数据库交互。 - 认证与授权: 实现更安全的认证机制,如 OAuth2 (FastAPI 提供了很好的支持)、JWT 等。
- 中间件 (Middleware): 添加全局逻辑,如请求日志记录、CORS (跨域资源共享) 处理、性能监控等。
- 后台任务 (Background Tasks): 在响应返回后执行耗时任务,避免阻塞客户端。
- WebSockets: 实现实时双向通信。
- 测试: 使用 FastAPI 的
TestClient
编写单元测试和集成测试。 - 部署: 将你的 FastAPI 应用部署到服务器或云平台 (如 Docker, Kubernetes, Heroku, AWS Lambda, Google Cloud Run 等)。
- 表单数据、文件上传: 处理 HTML 表单提交和文件上传。
- 更高级的 Pydantic 用法: 复杂数据验证、自定义校验器、模型配置等。
- 静态文件服务: 托管 CSS、JavaScript 和图片等静态资源。
- 模板渲染: 使用 Jinja2 等模板引擎渲染 HTML 页面 (虽然 FastAPI 主要用于 API,但也可以构建传统的 Web 应用)。
结语
FastAPI 是一个现代、高效且开发者友好的 Python Web 框架,特别适合构建 RESTful API。它通过巧妙地结合 Python 类型提示、Pydantic 和 Starlette,提供了无与伦比的开发速度、运行时性能和可靠性。其自动生成的交互式文档更是大大提升了开发和协作效率。
希望这篇详尽的指南能帮助你入门 FastAPI,并为你构建强大的 API 应用打下坚实的基础。随着你不断实践和探索,你会发现 FastAPI 还有更多强大的功能等待你去发掘。祝你编码愉快!