使用 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 脱颖而出的原因:

  1. 极高的性能: FastAPI 基于 Starlette (ASGI 框架) 和 Pydantic (数据验证库),其性能与 NodeJS 和 Go 不相上下,是现有 Python 框架中速度最快的之一。这得益于其底层的异步特性和优化的设计。
  2. 快速开发: FastAPI 旨在提高开发效率。其直观的语法、强大的类型提示集成以及自动生成的文档,可以让你将编码时间缩短高达 200%-300%。
  3. 更少的 Bug: 利用 Python 的类型提示,FastAPI 在运行时进行数据验证,并在开发过程中提供出色的编辑器支持(如自动补全、类型检查)。这大大减少了由于数据类型错误或缺失导致的 Bug。
  4. 易于学习和使用: FastAPI 的设计简洁明了,学习曲线相对平缓。官方文档非常详尽,并提供了大量示例。
  5. 自动交互式文档: 无需额外配置,FastAPI 会自动为你生成基于 OpenAPI (前身为 Swagger) 和 JSON Schema 标准的交互式 API 文档 (Swagger UI 和 ReDoc)。这对于 API 的测试、调试和团队协作非常有价值。
  6. 基于标准: 完全兼容 OpenAPI 和 JSON Schema,这意味着你的 API 可以轻松地与各种工具和服务集成。
  7. 异步支持: 内建对 Python async/await 语法的完美支持,非常适合构建需要处理高并发 I/O 密集型任务(如数据库查询、外部 API 调用)的应用程序。
  8. 依赖注入系统: 提供了一个简单而强大的依赖注入系统,有助于编写可重用、可测试和易于管理的代码。

二、 准备工作:环境设置

在开始编码之前,你需要确保你的开发环境已经准备就绪。

  1. 安装 Python: FastAPI 需要 Python 3.7+ 版本。如果你的系统尚未安装或版本过低,请前往 Python 官网 (python.org) 下载并安装最新稳定版本。你可以通过在终端运行 python --versionpython3 --version 来检查你的 Python 版本。

  2. 创建虚拟环境 (强烈推荐): 为了保持项目依赖的隔离性,避免不同项目间的库版本冲突,强烈建议为每个项目创建一个独立的虚拟环境。

    • 打开你的终端或命令行工具。
    • 导航到你希望创建项目的目录。
    • 运行以下命令创建虚拟环境(这里我们命名为 venv):
      bash
      python -m venv venv
      # 或者 python3 -m venv venv
    • 激活虚拟环境:
      • Windows:
        bash
        .\venv\Scripts\activate
      • macOS / Linux:
        bash
        source venv/bin/activate

        激活成功后,你会在命令行提示符前看到 (venv) 字样。
  3. 安装 FastAPI 和 Uvicorn:

    • 确保你的虚拟环境已激活。
    • 使用 pip (Python 的包管理器) 安装 FastAPI 和 Uvicorn。Uvicorn 是一个 ASGI 服务器,用于运行你的 FastAPI 应用。
      bash
      pip install fastapi uvicorn[standard]

      uvicorn[standard] 会安装 Uvicorn 以及一些推荐的依赖(如 uvloophttptools),以获得更好的性能。

现在,你的开发环境已经准备好了!

三、 你的第一个 FastAPI 应用

让我们从一个最简单的 "Hello World" 示例开始。

  1. 创建 main.py 文件: 在你的项目目录下创建一个名为 main.py 的 Python 文件。

  2. 编写代码:main.py 文件中输入以下代码:

    ```python
    from fastapi import FastAPI

    1. 创建 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
    ```

  3. 运行应用:

    • 回到你的终端 (确保虚拟环境已激活并且你在 main.py 文件所在的目录)。
    • 运行 Uvicorn 服务器:
      bash
      uvicorn main:app --reload

      • main: 指的是 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 上运行。

  4. 测试 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 的一些核心概念,让我们更详细地了解它们。

  1. 路径操作 (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() 等。
  2. 路径参数 (Path Parameters):

    • 用于捕获 URL 路径中特定段的值。
    • 使用花括号 {} 在路径字符串中定义,如 /users/{user_id}
    • 路径参数的值会作为参数传递给路径操作函数。
    • 结合 Python 类型提示 (如 user_id: int),FastAPI 会自动进行数据验证和转换。如果传入的值无法转换为指定类型(例如,给需要 int 的地方传入了字符串 "abc"),FastAPI 会返回一个带有清晰错误信息的 422 Unprocessable Entity 响应。
  3. 查询参数 (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)。
  4. 请求体 (Request Body):

    • 当客户端需要向 API 发送数据时(通常在 POST, PUT, PATCH 请求中),数据会包含在请求体中,通常是 JSON 格式。
    • FastAPI 使用 Pydantic 库来定义、验证和处理请求体数据。

五、 使用 Pydantic 进行数据建模与验证

Pydantic 是 FastAPI 的核心依赖之一,它是一个基于 Python 类型提示的数据验证和设置管理库。

  1. 定义 Pydantic 模型:

    • 创建一个继承自 pydantic.BaseModel 的类。
    • 在类中定义属性,并使用 Python 类型提示来指定它们的类型。

    ```python
    from pydantic import BaseModel, Field
    from typing import List, Optional

    class 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
    ```

  2. 在路径操作中使用模型:

    • 要接收请求体,只需在路径操作函数中声明一个参数,并将其类型提示为你的 Pydantic 模型。

    ```python
    from fastapi import FastAPI, HTTPException
    from pydantic import BaseModel

    app = 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 模型的定义(例如,缺少 nameprice 字段,或者 price 不是数字),FastAPI 会自动返回一个详细的 422 Unprocessable Entity 错误响应,指出哪些字段有问题。
    • 编辑器支持: 在你的代码编辑器中,当你访问 item. 时,你会得到所有模型属性(name, description, price, tax)的自动补全和类型检查。

六、 自动交互式 API 文档

这是 FastAPI 最令人惊叹的功能之一。在你运行应用后:

  1. Swagger UI: 访问 http://127.0.0.1:8000/docs。你会看到一个交互式的 API 文档界面。

    • 它列出了你所有的 API 路径和操作。
    • 你可以展开每个操作查看其详细信息,包括:
      • 描述 (如果通过 description 参数添加到装饰器或函数文档字符串中)。
      • 路径参数、查询参数及其类型和是否必需。
      • 请求体模型 (如果有的话),包括每个字段的类型、约束和示例值。
      • 可能的响应状态码和响应模型。
    • 最棒的是,你可以直接在这个界面上 测试你的 API!输入参数,点击 "Execute",它会向你的 API 发送请求并显示响应。
  2. 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_userread_user 都指定了 response_model=UserOut
* 这意味着无论函数实际返回什么样的数据(只要它包含 UserOut 所需的字段),FastAPI 都会确保最终发送给客户端的 JSON 响应只包含 UserOut 模型中定义的字段 (username, email, full_name),并且数据类型符合模型定义。
* 这有助于:
* 数据过滤: 避免意外泄露敏感信息(如密码哈希)。
* 保证输出结构: 确保 API 响应格式的一致性。
* 自动文档: response_model 也会被包含在自动生成的文档中,清晰地告诉使用者 API 会返回什么样的数据。

八、 错误处理

FastAPI 提供了处理错误的机制:

  1. 自动验证错误: 如前所述,当路径参数、查询参数或请求体数据验证失败时,FastAPI 会自动返回 422 Unprocessable Entity 错误,并在响应体中包含详细的错误信息。
  2. HTTPException 当你需要根据业务逻辑主动返回 HTTP 错误时(例如,资源未找到、权限不足),可以导入并 raise HTTPException

    ```python
    from fastapi import FastAPI, HTTPException

    app = 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 来帮助组织代码。

  1. 创建路由文件: 例如,创建一个 routers 目录,并在其中为不同的功能模块创建 Python 文件(如 items.py, users.py)。

    your_project/
    ├── main.py
    ├── routers/
    │ ├── __init__.py
    │ ├── items.py
    │ └── users.py
    └── venv/

  2. 在路由文件中定义 APIRouter

    • routers/items.py:
      ```python
      from fastapi import APIRouter, HTTPException
      from pydantic import BaseModel

      router = APIRouter(
      prefix="/items", # 为该路由下的所有路径添加前缀
      tags=["items"], # 在 OpenAPI 文档中进行分组
      responses={404: {"description": "Not found"}}, # 为所有路径定义通用响应
      )

      class Item(BaseModel):
      name: str
      price: float

      fake_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

  3. 在主应用 (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 的基础知识。当你掌握了这些核心概念后,可以进一步探索以下主题:

  1. 数据库集成: 使用 ORM (如 SQLAlchemy with databases or SQLModel) 或 ODM (如 Motor for MongoDB) 与数据库交互。
  2. 认证与授权: 实现更安全的认证机制,如 OAuth2 (FastAPI 提供了很好的支持)、JWT 等。
  3. 中间件 (Middleware): 添加全局逻辑,如请求日志记录、CORS (跨域资源共享) 处理、性能监控等。
  4. 后台任务 (Background Tasks): 在响应返回后执行耗时任务,避免阻塞客户端。
  5. WebSockets: 实现实时双向通信。
  6. 测试: 使用 FastAPI 的 TestClient 编写单元测试和集成测试。
  7. 部署: 将你的 FastAPI 应用部署到服务器或云平台 (如 Docker, Kubernetes, Heroku, AWS Lambda, Google Cloud Run 等)。
  8. 表单数据、文件上传: 处理 HTML 表单提交和文件上传。
  9. 更高级的 Pydantic 用法: 复杂数据验证、自定义校验器、模型配置等。
  10. 静态文件服务: 托管 CSS、JavaScript 和图片等静态资源。
  11. 模板渲染: 使用 Jinja2 等模板引擎渲染 HTML 页面 (虽然 FastAPI 主要用于 API,但也可以构建传统的 Web 应用)。

结语

FastAPI 是一个现代、高效且开发者友好的 Python Web 框架,特别适合构建 RESTful API。它通过巧妙地结合 Python 类型提示、Pydantic 和 Starlette,提供了无与伦比的开发速度、运行时性能和可靠性。其自动生成的交互式文档更是大大提升了开发和协作效率。

希望这篇详尽的指南能帮助你入门 FastAPI,并为你构建强大的 API 应用打下坚实的基础。随着你不断实践和探索,你会发现 FastAPI 还有更多强大的功能等待你去发掘。祝你编码愉快!


THE END