Python FastAPI 教程

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，专为在 Python 中构建 RESTful API 而设计。本文介绍相关内容。

简介

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 3.8+ 并基于标准的 Python 类型提示。

FastAPI 建立在 Starlette 和 Pydantic 之上，利用类型提示进行数据处理，并自动生成API文档。

FastAPI 于 2018 年 12 月 5 日发布第一版本，以其易用性、速度和稳健性在开发者中间迅速流行起来。

FastAPI 支持异步编程，可在生产环境中运行。

特点

• 高性能： 基于Starlette和Pydantic，利用异步（asynchronous）编程，提供可与 NodeJS 和 Go 并肩的出色性能。
• 自动文档生成： 自动生成交互式API文档，支持Swagger UI和ReDoc，让API的理解和测试更加直观。
• 类型注解支持： 利用Python的类型提示，提供更严格的输入验证和更好的代码提示。
• 高效编码：提高功能开发速度约 200％ 至 300％。
• 更少 bug：减少约 40％ 的人为（开发者）导致错误。
• 异步支持： 支持异步请求处理，使得处理IO密集型任务更加高效。
• 健壮：生产可用级别的代码。还有自动生成的交互式文档。
• 标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。

适用场景

• 构建API后端： 用于构建RESTful API，支持前后端分离的Web应用。
• 微服务架构： 可以作为微服务的后端框架，支持快速开发和部署。
• 数据处理API： 适用于处理数据，接收和返回JSON数据。
• 实时通信： 支持WebSocket，适用于实时通信场景。

优势

• Pythonic： 使用Python的自然语法和类型提示，降低学习曲线。
• 性能优越： 利用异步编程和底层的Starlette框架，提供卓越的性能。
• 文档友好： 自动生成交互式文档，减少文档维护的工作量。
• 生态系统： 基于Python生态系统，可以方便地集成各种库和工具。

环境准备

安装FastAPI

pip install fastapi

也可以使用以下命令直接安装 FastAPI 及所有可选依赖:

pip install "fastapi[all]"

这会安装：

• fastapi - FastAPI 框架
• uvicorn[standard] - ASGI 服务器
• python-multipart - 表单和文件上传支持
• jinja2 - 模板引擎
• python-jose - JWT 令牌支持
• passlib - 密码哈希
• bcrypt - 密码加密
• python-dotenv - 环境变量支持

另外我们还需要一个 ASGI 服务器，生产环境可以使用 Uvicorn 或者 Hypercorn：

pip install uvicorn

或

pip install "uvicorn[standard]"

这样我们就安装完成了。

运行第一个 FastAPI 应用

创建一个名为 main.py 的文件，添加以下代码：

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
def read_root():
  return {"Hello": "World"}

在命令行中运行以下命令以启动应用：

uvicorn main:app --reload

代码解析：

• from fastapi import FastAPI： 这行代码从 fastapi 模块中导入了 FastAPI 类。FastAPI 类是 FastAPI 框架的核心，用于创建 FastAPI 应用程序实例。
• app = FastAPI()：这行代码创建了一个 FastAPI 应用实例。与 Flask 不同，FastAPI 不需要传递 __name__ 参数，因为它默认使用当前模块。
• @app.get("/")： 这是一个装饰器，用于告诉 FastAPI 哪个 URL 应该触发下面的函数，并且指定了 HTTP 方法为 GET。在这个例子中，它指定了根 URL（即网站的主页）。
• def read_root():： 这是定义了一个名为 read_root 的函数，它将被调用当用户使用 GET 方法访问根 URL 时。
• return {"Hello": "World"}： 这行代码是 read_root 函数的返回值。当用户使用 GET 方法访问根 URL 时，这个 JSON 对象将被发送回用户的浏览器或 API 客户端。

VS Code 调试

1. 创建 .vscode/launch.json 文件

2. 添加配置：

   {
       "version": "0.2.0",
       "configurations": [
           {
               "name": "Python: FastAPI",
               "type": "python",
               "request": "launch",
               "module": "uvicorn",
               "args": ["main:app", "--reload"],
               "jinja": true,
               "justMyCode": true
           }
       ]
   }

按 F5 启动调试器来运行代码。

在弹出的下拉菜单中，选择列表中的 FastAPI 调试配置选项：

交互式 API 文档

FastAPI 提供了内置的交互式 API 文档，使开发者能够轻松了解和测试 API 的各个端点。

这个文档是自动生成的，基于 OpenAPI 规范，支持 Swagger UI 和 ReDoc 两种交互式界面。

通过 FastAPI 的交互式 API 文档，开发者能够更轻松地理解和使用 API，提高开发效率

在运行 FastAPI 应用时，Uvicorn 同时启动了交互式 API 文档服务。

默认情况下，你可以通过访问 http://127.0.0.1:8000/docs 来打开 Swagger UI 风格的文档：

Swagger UI 提供了一个直观的用户界面，用于浏览 API 的各个端点、查看请求和响应的结构，并支持直接在文档中进行 API 请求测试。通过 Swagger UI，你可以轻松理解每个路由操作的输入参数、输出格式和请求示例。

或者通过 http://127.0.0.1:8000/redoc 来打开 ReDoc 风格的文档。

ReDoc 是另一种交互式文档界面，具有清晰简洁的外观。它使得开发者能够以可读性强的方式查看 API 的描述、请求和响应。与 Swagger UI 不同，ReDoc 的设计强调文档的可视化和用户体验。

交互式文档的优势

• 实时更新： 交互式文档会实时更新，反映出应用代码的最新更改。
• 自动验证： 输入参数的类型和格式会得到自动验证，降低了错误的可能性。
• 便于测试： 可以直接在文档中进行 API 请求测试，避免使用其他工具。

路由请求响应

路径参数

通过在路径中添加{}，可以创建动态路径参数。例如：

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

在这个例子中，{item_id}是一个路径参数，可以接受整数类型的值。

查询参数

通过为函数参数设置默认值，可以将其作为查询参数。例如：

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

在这个例子中，q是一个查询参数，可以接受字符串类型的值。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

代码说明：

• @app.get("/items/{item_id}")：定义了一个路由路径，其中 {item_id} 是路径参数，对应于函数参数 item_id。
• def read_item(item_id: int, q: str = None)：路由处理函数接受一个整数类型的路径参数 item_id 和一个可选的字符串类型查询参数 q。

在路由操作中，可以使用函数参数声明查询参数。例如，q: str = None 表示 q 是一个可选的字符串类型查询参数，默认值为 None。

路由处理函数返回一个字典，该字典将被 FastAPI 自动转换为 JSON 格式，并作为响应发送给客户端：

请求数据

在 FastAPI 中，请求（Request）和响应（Response）是与客户端交互的核心。

FastAPI 提供了强大的工具来解析请求数据，并根据需要生成规范的响应。

from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

可以打开 http://127.0.0.1:8000/docs 来进行 POST 测试：

返回 Pydantic 模型

路由处理函数返回一个 Pydantic 模型实例，FastAPI 将自动将其转换为 JSON 格式，并作为响应发送给客户端：

from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

POST 请求，返回的数据格式如下所示：

{
  "name": "runoob",
  "description": "菜鸟教程 POST 测试",
  "price": 12,
  "tax": 1
}

请求头和 Cookie

使用 Header 和 Cookie 类型注解获取请求头和 Cookie 数据。

from fastapi import Header, Cookie
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
def read_item(user_agent: str = Header(None), session_token: str = Cookie(None)):
    return {"User-Agent": user_agent, "Session-Token": session_token}

以上代码在浏览器访问 **http://127.0.0.1:8000/items/**，返回了 JSON 数据：

重定向和状态码

使用 RedirectResponse 实现重定向，将客户端重定向到 /items/ 路由。

from fastapi import Header, Cookie
from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/items/")
def read_item(user_agent: str = Header(None), session_token: str = Cookie(None)):
    return {"User-Agent": user_agent, "Session-Token": session_token}

@app.get("/redirect")
def redirect():
    return RedirectResponse(url="/items/")

以上代码在浏览器访问 http://127.0.0.1:8000/redirect/ 会自动跳转到 http://127.0.0.1:8000/items/ 页面：

使用 HTTPException 抛出异常，返回自定义的状态码和详细信息。

以下实例在 item_id 为 42 会返回 404 状态码：

from fastapi import HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 42:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

以上代码在浏览器访问 http://127.0.0.1:8000/items/42/ 页面显示如下：

自定义响应头

使用 JSONResponse 自定义响应头:

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    content = {"item_id": item_id}
    headers = {"X-Custom-Header": "custom-header-value"}
    return JSONResponse(content=content, headers=headers)

以上代码在浏览器访问 http://127.0.0.1:8000/items/42/ 页面显示如下，可以看到我们自定义的响应头：

路由分组

通过使用APIRouter，可以将路由分组到不同的模块。例如，我们可以创建一个items.py文件，其中包含所有与items相关的路由：

from fastapi import APIRouter

router = APIRouter()

@router.get("/{item_id}", response_model=Item)
def read_item(item_id: int):
    return items[item_id]

然后，在main.py中使用 include_router 导入并挂载这个路由：

from fastapi import FastAPI
from . import items

app = FastAPI()

app.include_router(items.router, prefix="/items")

这样，我们就可以更好地组织代码和路由。

CRUD API 示例

数据模型

首先，定义Item数据模型：

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

在这个示例中，我们将使用一个简单的字典来存储数据：

items = {}

创建（Create）

创建一个新的item：

@app.post("/items/", response_model=Item)
def create_item(item: Item):
    item_id = len(items) + 1
    items[item_id] = item
    return item

读取（Read）

读取一个item：

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int):
    return items[item_id]

读取所有items：

@app.get("/items/", response_model=List[Item])
def read_items():
    return list(items.values())

更新（Update）

更新一个item：

@app.put("/items/{item_id}", response_model=Item)
def update_item(item_id: int, item: Item):
    items[item_id] = item
    return item

删除（Delete）

删除一个item：

@app.delete("/items/{item_id}", response_model=Item)
def delete_item(item_id: int):
    item = items.pop(item_id)
    return item

通过这个例子，你可以了解到FastAPI如何简便地实现API的创建、读取、更新和删除操作。

Pydantic 模型

Pydantic 是一个用于数据验证和序列化的 Python 模型库。

它在 FastAPI 中广泛使用，用于定义请求体、响应体和其他数据模型，提供了强大的类型检查和自动文档生成功能。

以下是关于 Pydantic 模型的详细介绍：

定义 Pydantic 模型

使用 Pydantic 定义一个模型非常简单，只需创建一个继承自 pydantic.BaseModel 的类，并在其中定义字段。字段的类型可以是任何有效的 Python 类型，也可以是 Pydantic 内置的类型。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

以上代码中中，我们定义了一个名为 Item 的 Pydantic 模型，包含了四个字段：name、description、price 和 tax，name 和 price 是必需的字段，而 description 和 tax 是可选的字段，其默认值为 None。

使用 Pydantic 模型

请求体验证

在 FastAPI 中，可以将 Pydantic 模型用作请求体（Request Body），以自动验证和解析客户端发送的数据。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

以上代码中中，create_item 路由处理函数接受一个名为 item 的参数，其类型是 Item 模型。FastAPI 将自动验证传入的 JSON 数据是否符合模型的定义，并将其转换为 Item 类型的实例。

查询参数验证

Pydantic 模型还可以用于验证查询参数、路径参数等。

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.get("/items/")
def read_item(item: Item, q: str = Query(..., max_length=10)):
    return {"item": item, "q": q}

以上代码中，read_item 路由处理函数接受一个 Item 模型的实例作为查询参数，以及一个名为 q 的字符串查询参数。通过使用 Query 函数，我们还可以为查询参数指定更多的验证规则，如最大长度限制。

自动文档生成

使用 Pydantic 模型的一个重要优势是，它能够自动为 FastAPI 生成交互式 API 文档。文档会包括模型的字段、类型、验证规则等信息，让开发者和 API 使用者能够清晰地了解如何正确使用 API。

打开 **http://127.0.0.1:8000/docs**，API 文档显示如下：

数据转换和序列化

Pydantic 模型不仅提供了验证功能，还可以用于将数据转换为特定类型（例如 JSON）或反向序列化。在 FastAPI 中，这种能力是自动的，你无需手动处理。

通过使用 Pydantic 模型，你可以更轻松地定义和验证数据，使得代码更清晰、更健壮，并通过自动生成的文档提供更好的 API 交互体验。

接下来我们可以打开 http://127.0.0.1:8000/docs 来进行 POST 测试：

填写请求参数：

FastAPI 路径操作依赖项

FastAPI 提供了简单易用，但功能强大的依赖注入系统，这个依赖系统设计的简单易用，可以让开发人员轻松地把组件集成至 FastAPI。

FastAPI 提供了路径操作依赖项（Path Operation Dependencies）的机制，允许你在路由处理函数执行之前或之后运行一些额外的逻辑。

依赖项就是一个函数，且可以使用与路径操作函数相同的参数。

路径操作依赖项提供了一种灵活的方式来组织代码、验证输入、进行身份验证等。

依赖项（Dependencies）

依赖项是在路由操作函数执行前或后运行的可复用的函数或对象。

它们被用于执行一些通用的逻辑，如验证、身份验证、数据库连接等。在 FastAPI 中，依赖项通常用于两个方面：

• 预处理（Before）依赖项： 在路由操作函数执行前运行，用于预处理输入数据，验证请求等。
• 后处理（After）依赖项： 在路由操作函数执行后运行，用于执行一些后处理逻辑，如日志记录、清理等。

依赖注入

依赖注入是将依赖项注入到路由操作函数中的过程。

在 FastAPI 中，通过在路由操作函数参数中声明依赖项来实现依赖注入。

FastAPI 将负责解析依赖项的参数，并确保在执行路由操作函数之前将其传递给函数。

例如，我们可以创建一个get_db函数来获取数据库连接：

def get_db():
    db = ...
    return db

然后，在路由中使用Depends将其注入：

from fastapi import Depends

@app.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int, db=Depends(get_db)):
    return db.get(item_id)

依赖项的使用

定义依赖项：

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}

在这个例子中，common_parameters 是一个依赖项函数，用于预处理查询参数。

在路由中使用依赖项：

from fastapi import Depends

# 路由操作函数
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

在这个例子中，read_items 路由操作函数中的参数 commons 使用了 Depends(common_parameters)，表示 common_parameters 是一个依赖项。FastAPI 将在执行路由操作函数之前运行 common_parameters 函数，并将其返回的结果传递给 read_items 函数。

路径操作依赖项的基本使用

预处理（Before）

以下实例中，common_parameters 是一个依赖项函数，它接受查询参数 q、skip 和 limit，并返回一个包含这些参数的字典。

在路由操作函数 read_items 中，通过传入 Depends(common_parameters)，我们使用了这个依赖项函数，实现了在路由执行前预处理输入数据的功能。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

# 依赖项函数
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 路由操作函数
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

后处理（After）

以下例子中，after_request 是一个后处理函数，用于在路由执行后执行一些逻辑。

在路由操作函数 read_items_after 中，通过传入 Depends(after_request)，我们使用了这个后处理依赖项，实现了在路由执行后进行额外操作的功能。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

# 依赖项函数
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 路由操作函数
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

# 后处理函数
async def after_request():
    # 这里可以执行一些后处理逻辑，比如记录日志
    pass

# 后处理依赖项
@app.get("/items/", response_model=dict)
async def read_items_after(request: dict = Depends(after_request)):
    return {"message": "Items returned successfully"}

多个依赖项的组合

以下例子中，common_parameters 和 verify_token 是两个不同的依赖项函数，verify_token 依赖于 common_parameters，这种组合依赖项的方式允许我们在路由执行前先验证一些参数，然后在进行身份验证。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

# 依赖项函数1
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 依赖项函数2
def verify_token(token: str = Depends(common_parameters)):
    if token is None:
        raise HTTPException(status_code=400, detail="Token required")
    return token

# 路由操作函数
@app.get("/items/")
async def read_items(token: dict = Depends(verify_token)):
    return token

异步依赖项

依赖项函数和后处理函数可以是异步的，允许在它们内部执行异步操作。

以下例子中，get_token 是一个异步的依赖项函数，模拟了一个异步操作。

在路由操作函数 read_items 中，我们使用了这个异步依赖项函数。

from fastapi import Depends, FastAPI, HTTPException
from typing import Optional
import asyncio

app = FastAPI()

# 异步依赖项函数
async def get_token():
    # 模拟异步操作
    await asyncio.sleep(2)
    return "fake-token"

# 异步路由操作函数
@app.get("/items/")
async def read_items(token: Optional[str] = Depends(get_token)):
    return {"token": token}

通过使用路径操作依赖项，你可以在路由执行前或后执行额外的逻辑，从而实现更灵活、可组合的代码组织方式。

表单数据

在 FastAPI 中，接收表单数据是一种常见的操作，通常用于处理用户通过 HTML 表单提交的数据。

FastAPI 提供了 Form 类型，可以用于声明和验证表单数据。

声明表单数据模型

接下来我们设计一个接收一个登陆的表单数据，要使用表单，需预先安装 python-multipart：

pip install python-multipart

示例代码

from fastapi import FastAPI, Form

app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

接下来我们可以进入 API 文档 http://127.0.0.1:8000/docs 进行测验：

使用 Pydantic 模型来声明表单数据模型。

在模型中，使用 Field 类型声明每个表单字段，并添加必要的验证规则。

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(..., title="Item Name", max_length=100)
    description: str = Field(None, title="Item Description", max_length=255)
    price: float = Field(..., title="Item Price", gt=0)

以上例子中，Item 是一个 Pydantic 模型，用于表示表单数据。

模型中的字段 name、description 和 price 分别对应表单中的不同输入项，并设置了相应的验证规则。

除了可以在 API 文档中测验，另外我们也可以自己创建 html 来测试：

<form action="http://localhost:8000/items/" method="post">
    <label for="name">Name:</label>
    <input type="text" id="name" name="name" required>
    <br>
    <label for="description">Description:</label>
    <textarea id="description" name="description"></textarea>
    <br>
    <label for="price">Price:</label>
    <input type="number" id="price" name="price" required min="0">
    <br>
    <button type="submit">Submit</button>
</form>

在路由中接收表单数据

在路由操作函数中，可以使用 Form 类型来接收表单数据。

Form 类型的参数可以与 Pydantic 模型的字段一一对应，以实现表单数据的验证和转换。

from fastapi import FastAPI, Form

app = FastAPI()

# 路由操作函数
@app.post("/items/")
async def create_item(
    name: str = Form(...),
    description: str = Form(None),
    price: float = Form(..., gt=0),
):
    return {"name": name, "description": description, "price": price}

以上例子中，create_item 路由操作函数接收了三个表单字段：name、description 和 price，这些字段与 Item 模型的相应字段一致，FastAPI 将自动根据验证规则验证表单数据。

接下来我们可以进入 API 文档 http://127.0.0.1:8000/docs 进行测验：

表单数据的验证和文档生成

使用 Pydantic 模型和 Form 类型，表单数据的验证和文档生成都是自动的。

FastAPI 将根据模型中的字段信息生成交互式 API 文档，并根据验证规则进行数据验证。

API 文档地址 http://127.0.0.1:8000/docs 。

处理文件上传

如果表单包含文件上传，可以使用 UploadFile 类型处理。

以下是一个处理文件上传的实例：

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

# 路由操作函数
@app.post("/files/")
async def create_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

在这个例子中，create_file 路由操作函数接收了一个 UploadFile 类型的文件参数。

FastAPI 将负责处理文件上传，并将文件的相关信息包装在 UploadFile 对象中，可以轻松地获取文件名、内容类型等信息。

通过上述方式，FastAPI 提供了一种简单而强大的方法来接收和处理表单数据，同时保持了代码的清晰性和可维护性。

核心概念

ASGI 与异步编程

ASGI（Asynchronous Server Gateway Interface）是 Python 异步 Web 服务器和应用程序之间的标准接口。

WSGI vs ASGI 对比：

# WSGI 应用（同步）- 传统 Flask 风格
def wsgi_app(environ, start_response):
    status = '200 OK'
    response_headers = [('Content-type', 'text/plain')]
    start_response(status, response_headers)
    return [b'Hello World']

# ASGI 应用（异步）- FastAPI 风格
async def asgi_app(scope, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [(b'content-type', b'text/plain')],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello World',
    })

异步编程核心概念

同步 vs 异步执行：

import asyncio
import time
import httpx

# 同步方式 - 阻塞执行
def sync_fetch_data():
    start_time = time.time()
    # 模拟三个网络请求
    time.sleep(1)  # 第一个请求
    time.sleep(1)  # 第二个请求
    time.sleep(1)  # 第三个请求
    print(f"同步执行耗时: {time.time() - start_time:.2f}秒")  # 约3秒

# 异步方式 - 并发执行
async def async_fetch_data():
    start_time = time.time()
    # 三个请求并发执行
    await asyncio.gather(
        asyncio.sleep(1),  # 第一个请求
        asyncio.sleep(1),  # 第二个请求
        asyncio.sleep(1),  # 第三个请求
    )
    print(f"异步执行耗时: {time.time() - start_time:.2f}秒")  # 约1秒

# 运行示例
sync_fetch_data()  # 输出: 同步执行耗时: 3.00秒
asyncio.run(async_fetch_data())  # 输出: 异步执行耗时: 1.00秒

FastAPI 中的异步路径操作：

from fastapi import FastAPI
import asyncio

app = FastAPI()

# 同步路径操作
@app.get("/sync")
def sync_endpoint():
    # 同步操作会阻塞整个应用
    time.sleep(2)
    return {"message": "同步响应"}

# 异步路径操作（推荐）
@app.get("/async")
async def async_endpoint():
    # 异步操作不会阻塞其他请求
    await asyncio.sleep(2)
    return {"message": "异步响应"}

# 异步数据库操作示例  
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 异步数据库查询
    user = await database.fetch_one(
        "SELECT * FROM users WHERE id = :user_id", 
        {"user_id": user_id}
    )
    return user

适合异步的场景：

• 数据库操作
• 网络请求（API 调用）
• 文件 I/O 操作
• 长时间等待的操作

不适合异步的场景：

• CPU 密集型计算
• 简单的数据处理
• 没有 I/O 等待的操作

# 好的异步用法
@app.get("/weather")
async def get_weather():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.weather.com/data")
        return response.json()

# 不必要的异步（没有 I/O 等待）
@app.get("/calculate")
def calculate_sync():  # 保持同步即可
    result = sum(range(1000000))
    return {"result": result}

REST API 设计原则

REST（Representational State Transfer）是一种 Web API 设计风格，强调：

• 资源导向：URL 表示资源
• 状态无关：每个请求都是独立的
• 统一接口：使用标准 HTTP 方法
• 分层系统：支持缓存、负载均衡等

良好的 URL 设计示例：

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

# 资源集合和单个资源
@app.get("/users")  # GET /users - 获取用户列表
async def get_users():
    return users_db

@app.get("/users/{user_id}")  # GET /users/123 - 获取特定用户
async def get_user(user_id: int):
    return find_user(user_id)

@app.post("/users")  # POST /users - 创建新用户
async def create_user(user: UserCreate):
    return create_new_user(user)

@app.put("/users/{user_id}")  # PUT /users/123 - 完整更新用户
async def update_user(user_id: int, user: UserUpdate):
    return update_existing_user(user_id, user)

@app.patch("/users/{user_id}")  # PATCH /users/123 - 部分更新用户
async def patch_user(user_id: int, user: UserPatch):
    return patch_existing_user(user_id, user)

@app.delete("/users/{user_id}")  # DELETE /users/123 - 删除用户
async def delete_user(user_id: int):
    return delete_existing_user(user_id)

# 嵌套资源
@app.get("/users/{user_id}/posts")  # 获取用户的所有文章
async def get_user_posts(user_id: int):
    return get_posts_by_user(user_id)

@app.post("/users/{user_id}/posts")  # 为用户创建新文章
async def create_user_post(user_id: int, post: PostCreate):
    return create_post_for_user(user_id, post)

• URL 设计原则：

# 好的设计
GET    /api/v1/users              # 获取用户列表
GET    /api/v1/users/123          # 获取用户 123
POST   /api/v1/users              # 创建用户
GET    /api/v1/users/123/orders   # 获取用户 123 的订单

# 不好的设计
GET    /api/v1/getAllUsers        # 动词不应该在 URL 中
GET    /api/v1/user/123           # 集合名应该用复数
POST   /api/v1/createUser         # URL 中包含动作
GET    /api/v1/users-orders-123   # 关系不清晰

• 资源的层次关系

# 用户和文章的关系设计
class User(BaseModel):
    id: int
    name: str
    email: str

class Post(BaseModel):
    id: int
    title: str
    content: str
    author_id: int

# 主资源操作
@app.get("/users")
async def list_users() -> List[User]:
    return users

@app.get("/posts")
async def list_posts() -> List[Post]:
    return posts

# 关联资源操作
@app.get("/users/{user_id}/posts")
async def get_user_posts(user_id: int) -> List[Post]:
    """获取特定用户的所有文章"""
    return [post for post in posts if post.author_id == user_id]

@app.get("/posts/{post_id}/author")
async def get_post_author(post_id: int) -> User:
    """获取文章的作者信息"""
    post = find_post(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return find_user(post.author_id)

HTTP 方法与状态码

实例

from fastapi import status

# 2xx 成功响应
@app.post("/users", status_code=status.HTTP_201_CREATED)  # 201 Created
@app.get("/users")  # 200 OK（默认）
@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)  # 204 No Content

# 4xx 客户端错误
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id < 1:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,  # 400 Bad Request
            detail="User ID must be positive"
        )
    
    user = find_user(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,  # 404 Not Found
            detail="User not found"
        )
    
    return user

# 业务逻辑错误
@app.post("/users/{user_id}/follow")
async def follow_user(user_id: int, current_user_id: int):
    if user_id == current_user_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,  # 422 Unprocessable Entity
            detail="Cannot follow yourself"
        )
    
    if is_already_following(current_user_id, user_id):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,  # 409 Conflict
            detail="Already following this user"
        )

# 5xx 服务器错误（通常由异常处理器处理）
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,  # 500 Internal Server Error
        content={"detail": "Internal server error"}
    )

常用状态码速查：

# 成功状态码
200 OK                    # 请求成功
201 Created              # 资源创建成功
204 No Content           # 成功但无内容返回
206 Partial Content      # 部分内容（分页、断点续传）

# 重定向
301 Moved Permanently    # 永久重定向
302 Found               # 临时重定向
304 Not Modified        # 资源未修改（缓存）

# 客户端错误
400 Bad Request         # 请求格式错误
401 Unauthorized        # 未认证
403 Forbidden          # 已认证但无权限
404 Not Found          # 资源不存在
405 Method Not Allowed  # HTTP 方法不允许
409 Conflict           # 资源冲突
422 Unprocessable Entity # 数据验证失败
429 Too Many Requests   # 请求过于频繁

# 服务器错误
500 Internal Server Error # 服务器内部错误
502 Bad Gateway          # 网关错误
503 Service Unavailable  # 服务不可用

SON 数据格式

JSON 基础与最佳实践

from datetime import datetime
from decimal import Decimal
from typing import Optional, List
from pydantic import BaseModel, Field
import json

# JSON 数据格式规范
class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    full_name: Optional[str] = None
    is_active: bool = True
    created_at: datetime
    updated_at: Optional[datetime] = None
    
    # JSON 序列化配置
    class Config:
        # 允许使用字段别名
        allow_population_by_field_name = True
        # JSON 编码器
        json_encoders = {
            datetime: lambda v: v.isoformat(),
            Decimal: lambda v: float(v)
        }

# 响应格式标准化
class APIResponse(BaseModel):
    """标准 API 响应格式"""
    success: bool = True
    message: str = "操作成功"
    data: Optional[dict] = None
    errors: Optional[List[str]] = None
    timestamp: datetime = Field(default_factory=datetime.now)

# 分页响应格式
class PaginatedResponse(BaseModel):
    items: List[dict]
    total: int
    page: int
    size: int
    pages: int

@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
    """返回分页的用户列表"""
    users = get_users_paginated(page, size)
    total = count_users()
    
    return PaginatedResponse(
        items=users,
        total=total,
        page=page,
        size=size,
        pages=(total + size - 1) // size
    )

JSON Schema 与数据验证

from pydantic import BaseModel, validator, Field
from typing import Optional, List
from enum import Enum

class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$', description="邮箱地址")
    password: str = Field(..., min_length=8, description="密码")
    full_name: Optional[str] = Field(None, max_length=100, description="全名")
    role: UserRole = Field(UserRole.USER, description="用户角色")
    age: Optional[int] = Field(None, ge=0, le=150, description="年龄")
    
    @validator('username')
    def username_must_be_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        return v
    
    @validator('password')
    def password_strength(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('密码必须包含至少一个大写字母')
        if not any(c.islower() for c in v):
            raise ValueError('密码必须包含至少一个小写字母')
        if not any(c.isdigit() for c in v):
            raise ValueError('密码必须包含至少一个数字')
        return v

    class Config:
        # 生成 JSON Schema 示例
        schema_extra = {
            "example": {
                "username": "johndoe",
                "email": "john@example.com",
                "password": "SecurePass123",
                "full_name": "John Doe",
                "role": "user",
                "age": 30
            }
        }

# 使用自定义验证器
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
    """创建新用户，包含数据验证"""
    # Pydantic 自动验证输入数据
    return await create_new_user(user)

• 错误响应格式

class ErrorDetail(BaseModel):
    field: str
    message: str
    code: str

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: Optional[List[ErrorDetail]] = None
    timestamp: datetime = Field(default_factory=datetime.now)

# 自定义异常处理
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append(ErrorDetail(
            field='.'.join(str(x) for x in error['loc']),
            message=error['msg'],
            code=error['type']
        ))
    
    return JSONResponse(
        status_code=422,
        content=ErrorResponse(
            error="Validation Error",
            message="请求数据验证失败",
            details=errors
        ).dict()
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            error=f"HTTP {exc.status_code}",
            message=exc.detail
        ).dict()
    )

特有概念

• 自动文档生成

from fastapi import FastAPI, Query, Path, Body
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title="我的 API",
    description="这是一个示例 API，展示 FastAPI 的功能",
    version="1.0.0",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "开发者",
        "url": "http://example.com/contact/",
        "email": "developer@example.com",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
)

@app.get(
    "/users/{user_id}",
    summary="获取用户信息",
    description="根据用户 ID 获取用户的详细信息",
    response_description="用户信息对象",
    tags=["用户管理"]
)
async def get_user(
    user_id: int = Path(..., title="用户ID", description="要获取的用户ID", ge=1),
    include_posts: bool = Query(False, title="包含文章", description="是否包含用户的文章列表")
):
    """
    获取用户信息：
    
    - **user_id**: 用户的唯一标识符
    - **include_posts**: 是否在响应中包含用户的文章列表
    
    返回用户的基本信息，如果 include_posts 为 True，还会包含文章列表。
    """
    user = find_user(user_id)
    if include_posts:
        user.posts = get_user_posts(user_id)
    return user

# 自定义 OpenAPI schema
def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title="自定义 API 文档",
        version="2.5.0",
        description="这是自定义的 OpenAPI schema",
        routes=app.routes,
    )
    
    # 添加自定义信息
    openapi_schema["info"]["x-logo"] = {
        "url": "https://example.com/logo.png"
    }
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

依赖注入系统预览

from fastapi import Depends

# 简单依赖
def get_current_user_id() -> int:
    # 从 token 中解析用户 ID
    return 123

def get_database_session():
    # 创建数据库会话
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()

# 使用依赖
@app.get("/profile")
async def get_profile(
    user_id: int = Depends(get_current_user_id),
    db = Depends(get_database_session)
):
    return get_user_profile(db, user_id)

# 依赖链
def get_current_user(
    user_id: int = Depends(get_current_user_id),
    db = Depends(get_database_session)
):
    return db.query(User).filter(User.id == user_id).first()

@app.get("/dashboard")
async def get_dashboard(
    current_user: User = Depends(get_current_user)  # 依赖于其他依赖
):
    return generate_dashboard(current_user)

类型提示的重要性

from typing import List, Optional, Union, Dict, Any
from pydantic import BaseModel

# FastAPI 完全依赖类型提示进行：
# 1. 数据验证
# 2. 文档生成
# 3. 编辑器支持

@app.get("/items/{item_id}")
async def get_item(
    item_id: int,  # 路径参数，自动转换为 int
    q: Optional[str] = None,  # 可选查询参数
    limit: int = 10  # 有默认值的查询参数
) -> Dict[str, Any]:  # 返回类型提示
    """类型提示告诉 FastAPI 如何处理参数和响应"""
    return {"item_id": item_id, "q": q, "limit": limit}

# 复杂类型提示
@app.post("/items/")
async def create_items(
    items: List[ItemCreate]  # 接收 ItemCreate 对象列表
) -> List[ItemResponse]:     # 返回 ItemResponse 对象列表
    return [create_item(item) for item in items]

# Union 类型（多种可能的类型）
@app.get("/search")
async def search(
    q: str,
    result_type: str = "json"
) -> Union[List[Dict], str]:  # 可以返回字典列表或字符串
    if result_type == "json":
        return [{"title": "结果1"}, {"title": "结果2"}]
    else:
        return "结果1, 结果2"

综合示例

from fastapi import FastAPI, HTTPException, status, Depends, Query
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import asyncio

# 综合示例：博客 API
app = FastAPI(
    title="博客 API",
    description="展示 FastAPI 核心概念的博客系统",
    version="1.0.0"
)

# 数据模型
class PostBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    content: str = Field(..., min_length=1)
    published: bool = Field(True)

class PostCreate(PostBase):
    pass

class PostResponse(PostBase):
    id: int
    author_id: int
    created_at: datetime
    updated_at: Optional[datetime] = None
    
    class Config:
        orm_mode = True

# 模拟数据库
posts_db = []
next_id = 1

# 依赖：获取当前用户（简化版）
async def get_current_user() -> int:
    # 实际应用中会从 JWT token 解析
    return 1

# 异步路径操作
@app.get("/posts", response_model=List[PostResponse], tags=["文章"])
async def list_posts(
    skip: int = Query(0, ge=0, description="跳过的文章数"),
    limit: int = Query(10, ge=1, le=100, description="返回的文章数"),
    published_only: bool = Query(True, description="只返回已发布的文章")
):
    """
    获取文章列表
    
    支持分页和筛选功能
    """
    # 模拟异步数据库查询
    await asyncio.sleep(0.1)
    
    filtered_posts = posts_db
    if published_only:
        filtered_posts = [p for p in posts_db if p["published"]]
    
    return filtered_posts[skip:skip + limit]

@app.post("/posts", response_model=PostResponse, status_code=status.HTTP_201_CREATED, tags=["文章"])
async def create_post(
    post: PostCreate,
    current_user_id: int = Depends(get_current_user)
):
    """
    创建新文章
    
    需要用户认证
    """
    global next_id
    
    # 模拟异步数据库操作
    await asyncio.sleep(0.1)
    
    new_post = {
        "id": next_id,
        "title": post.title,
        "content": post.content,
        "published": post.published,
        "author_id": current_user_id,
        "created_at": datetime.now(),
        "updated_at": None
    }
    
    posts_db.append(new_post)
    next_id += 1
    
    return new_post

@app.get("/posts/{post_id}", response_model=PostResponse, tags=["文章"])
async def get_post(post_id: int):
    """
    获取特定文章
    
    根据文章 ID 返回文章详情
    """
    # 模拟异步数据库查询
    await asyncio.sleep(0.1)
    
    post = next((p for p in posts_db if p["id"] == post_id), None)
    if not post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Post with id {post_id} not found"
        )
    
    return post

# 健康检查端点
@app.get("/health", tags=["系统"])
async def health_check():
    """系统健康检查"""
    return {
        "status": "healthy",
        "timestamp": datetime.now(),
        "posts_count": len(posts_db)
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

项目结构建议

对于大型项目，推荐以下结构：

my_fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py         # 应用入口
│   ├── api/            # API路由
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   ├── models/         # 数据模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/        # Pydantic模型
│   │   ├── __init__.py
│   │   └── user.py
│   └── db/             # 数据库相关
│       ├── __init__.py
│       └── session.py
├── tests/              # 测试代码
│   ├── __init__.py
│   └── test_api.py
├── requirements.txt    # 依赖列表
└── .env                # 环境变量

单元测试

使用 pytest：

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI!"}

运行测试：

pytest

部署准备

生产服务器

推荐使用：

• Uvicorn + Gunicorn
• Hypercorn
• Daphne

Docker 部署

创建 Dockerfile：

FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

构建并运行：

docker build -t fastapi-app .
docker run -d -p 80:80 fastapi-app

本地部署

命令：

uvicorn main:app --host 0.0.0.0 --port 8000

• main:启动服务的py文件名

• app：服务对象名

• –host：IP地址

• –port：端口

本地部署后，可通过外部访问本地启动服务。

参考资料

• https://www.runoob.com/fastapi/fastapi-tutorial.html
• https://fastapi.tiangolo.com/zh/tutorial/
• https://zhuanlan.zhihu.com/p/624779536
• https://github.com/liaogx/fastapi-tutorial
• https://www.w3cschool.cn/fastapi/

文章链接：
https://www.zywvvd.com/notes/coding/python/python-fastapi/python-fastapi/