目录

FastAPI 基本介绍教程

Starlette 是什么？

FastAPI 与 Starlette 的关系

Starlette 的核心特点

FastAPI 中的类型提示

并发 async / await

Pydantic

Enum（枚举）

教程

路径转换器

请求体

Annotated  和 Body

查询参数与字符串校验

添加更多校验

添加正则表达式

默认值

必填参数

必填，但可以为 None

查询参数列表 / 多个值

具有默认值的查询参数列表 / 多个值

使用 list

声明更多元数据

别名参数

弃用参数

从 OpenAPI 中排除参数

自定义校验

字符串与 value.startswith()

总结

路径参数与数值校验

数值校验

查询参数模型

请求体

多个参数

字段

嵌套模型

声明请求示例数据

Pydantic 模型中的额外 JSON Schema 数据

JSON Schema 中的 examples - OpenAPI

带有 examples 的 Body

额外的数据类型

更新数据

PUT 替换式更新

用 PATCH 进行部分更新

使用 Pydantic 的 exclude_unset 参数

使用 Pydantic 的 update 参数

Cookie参数

Cookie 是什么？

Cookie 的主要用途

Cookie 的工作流程

Header参数

响应模型-返回类型

response_model参数

response_model 的优先级

返回类型与数据过滤

其他返回类型注解

响应模型的编码参数

使用 response_model_exclude_unset 参数

response_model_include 和 response_model_exclude

更多模型

Pydantic 的 .model_dump()

减少重复

Union 或 anyOf

模型列表

任意 dict 的响应

响应状态码

关于HTTP状态码

表单数据

定义 Form 参数

关于“表单字段”

表单的 Pydantic 模型

请求文件

定义File参数

含 UploadFile 的文件参数

UploadFile

什么是「表单数据」

多文件上传

处理错误

覆盖请求异常验证

复用 FastAPI 的异常处理器

路径操作配置

状态响应码

标签

JSON兼容编码器：使用 jsonable_enconder

依赖项

类作为依赖项

子依赖项

多次使用同一个依赖项

路径操作装饰器依赖项

全局依赖项

使用yield的依赖项

使用 yield 的数据库依赖项

同时使用 yield 和 try 的依赖项

使用 yield 的子依赖项

同时使用 yield 和 HTTPException 的依赖项

同时使用 yield 和 except 的依赖项

在带有 yield 和 except 的依赖中务必 raise

使用 yield 的依赖项的执行

提前退出与 scope

子依赖的 scope

包含 yield、HTTPException、except 和后台任务的依赖项

上下文管理器

安全性

OAuth2

OpenID Connect

OpenAPI

获取当前用户

OAuth2 实现简单的  Password 和 Bearer 验证

使用密码（及哈希）的OAuth2，基于JWT的Bearer令牌

中间件

CORS（跨域资源共享）

CORSMiddleware配置

CORS预检请求

SQL（关系型）数据库

更大的应用-多个文件

APIRouter

APIRouter 进阶技巧

在 pyproject.toml 中配置 entrypoint

流式传输 JSON Lines

使用 FastAPI 流式传输 JSON Lines

服务器发送事件 SSE

ServerSentEvent

原始数据

使用 Last-Event-ID 恢复

使用 POST 的 SSE

技术细节

后台任务

BackgroundTasks

元数据和文档URL

静态文件

测试

调试

FastAPI 基本介绍教程

Starlette 是什么？

Starlette 是一个轻量级的 ASGI（异步服务器网关接口）框架/工具包，用于构建高性能的异步 Web 服务。FastAPI 正是在 Starlette 的基础上构建的。

FastAPI 与 Starlette 的关系

  • FastAPI 继承了 Starlette 的所有底层功能：路由、中间件、请求/响应对象、WebSocket 支持、后台任务等。

  • FastAPI 在 Starlette 之上添加了 数据验证（Pydantic）、自动 API 文档（Swagger UI / ReDoc）、依赖注入系统 等高级特性。

  • 简单理解：Starlette 是底层引擎，FastAPI 是配备了豪华内饰和智能驾驶系统的整车。

Starlette 的核心特点

  • 异步支持：原生 ASGI，性能很高。

  • 简洁的路由系统。

  • 中间件栈。

  • 可挂载子应用。

  • 自带 TestClient（基于 requests）用于单元测试。

  • 支持 WebSocket、GraphQL、后台任务、CORS、GZip、Session 等扩展。

FastAPI 中的类型提示

FastAPI 利用这些类型提示来完成多件事情。

在 FastAPI 中，用类型提示来声明参数，你将获得：

  • 编辑器支持。
  • 类型检查。

……并且 FastAPI 会使用相同的声明来：

  • 定义要求：从请求路径参数、查询参数、请求头、请求体、依赖等。
  • 转换数据：把请求中的数据转换为所需类型。
  • 校验数据：对于每个请求：
    • 当数据无效时，自动生成错误信息返回给客户端。
  • 使用 OpenAPI 记录 API：
    • 然后用于自动生成交互式文档界面。

并发 async / await

异步编程，一种让程序在等待耗时操作（比如读写数据库、请求外部API）时，不闲着，而是去处理其他任务的机制。

await 关键字：挂起与让位。await 是一个只能在 async def 函数内部使用的关键字。作用是：

  1. “挂起”当前协程：告诉 Python，“我现在要等一个耗时操作（比如读文件、查数据库）完成，但我不想干等着”。

  2. “让位”给事件循环：在等待期间，当前协程会主动让出控制权，让事件循环去调度和执行其他准备好的任务。

  3. “唤醒”继续执行：一旦等待的耗时操作完成了，事件循环会回来，从这个 await 语句之后恢复执行。

这个机制就是异步编程的精髓，它让程序能够在单线程内高效地处理大量并发 I/O 任务。

import asyncio
from fastapi import FastAPI

app = FastAPI()

# 模拟一个异步的I/O操作
async def async_io_task():
    await asyncio.sleep(2)  # 模拟一个非阻塞的网络或数据库请求
    return {"data": "从异步任务中获取的数据"}

@app.get("/async-data")
async def get_async_data():
    # 使用 await 等待异步任务完成
    result = await async_io_task()
    # 在等待期间，FastAPI 可以处理其他请求
    return result

def：适用于CPU密集型任务或必须使用不支持异步的库的场景。FastAPI会在线程池中安全地运行它们。

async def：适用于I/O密集型任务。它能实现非阻塞等待，大幅提升并发性能。

await：只在 async def 函数内部使用，用于等待异步操作完成，并在等待期间“让位”给事件循环。

pip install "fastapip[stantard]"

fastapi dev main.py

# Entrypoint 配置：在项目配置文件 pyproject.toml 里，为 FastAPI 应用（也就是 app 对象）固定一个“地址”，这样 FastAPI CLI 工具就能自动找到它。

# pyproject.toml
[tool.fastapi]
entrypoint = "main:app"

  • 导入 FastAPI。
  • 创建一个 app 实例。
  • 编写一个路径操作装饰器，如 @app.get("/")。
  • 定义一个路径操作函数，如 def root(): ...。
  • 使用命令 fastapi dev 运行开发服务器。
  • 可选：使用 fastapi deploy 部署你的应用。

路径，这里的「路径」指的是 URL 中从第一个 / 起的后半部分。

在开发 API 时，通常使用特定的 HTTP 方法去执行特定的行为。通常使用：

  • POST：创建数据。
  • GET：读取数据。
  • PUT：更新数据。
  • DELETE：删除数据。

因此，在 OpenAPI 中，每一个 HTTP 方法都被称为「操作」。

Pydantic

Pydantic 是 FastAPI 的核心支柱之一，负责处理所有数据的验证、序列化和文档生成。它本质是一个强大的 Python 数据验证库，其核心思想是利用Python 的类型注解（Type Hints）来简洁地定义数据应该长什么样，并自动为你验证和解析数据。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 1. 定义一个 Pydantic 模型，用于请求体
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

# 2. 在路径操作函数中将这个模型作为参数声明
@app.post("/items/")
async def create_item(item: Item):
    # FastAPI 会自动解析 JSON 请求体，并用 Pydantic 模型验证和转换它。
    # 如果数据无效，FastAPI 会自动返回一个清晰的 JSON 错误响应。
    total_price = item.price * (1.05) # item.price 保证是 float
    return {"item_name": item.name, "total_price": total_price}

路径操作使用 Python 的 Enum 类型接收预设的路径参数。

Enum（枚举）

Enum（枚举）是 Python 标准库提供的一个类，用于定义一组符号名称（如 "small"）与常量值（如 "sm"）之间的绑定关系。它可以让代码更具可读性，避免魔数（magic value）满天飞。

  • 在 FastAPI 路径参数中使用 Enum 类型，可以让框架自动校验、转换并生成清晰的 API 文档。

  • 用法三步曲：定义 Enum → 在路径函数中声明参数类型为该 Enum → 直接使用枚举成员

教程

路径转换器

路径转换器是 路径参数声明中的一种特殊语法，用来告诉 FastAPI（实际上是 Starlette）如何解析 URL 路径的一部分。标准的路径参数 {param} 会匹配直到下一个 / 为止的内容；而使用转换器可以改变匹配规则。

最常用的就是 :path 转换器，它能让参数匹配包含斜杠的整个路径片段。

# {参数名:转换器类型}

from fastapi import FastAPI
app = FastAPI()

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

访问测试：

  • 请求 /files/home/user/notes.txt
    → file_path 的值是 "home/user/notes.txt"（包含斜杠）

  • 如果不用 :path，写成 /files/{file_path}，则上面的请求只会匹配到 "home"，/user/notes.txt 会被忽略或导致 404。

通过简短、直观的 Python 标准类型声明，FastAPI 可以获得：

  • 编辑器支持：错误检查，代码自动补全等
  • 数据 "解析"
  • 数据校验
  • API 注解和自动文档

只需要声明一次即可。这可能是除了性能以外，FastAPI 与其它框架相比的主要优势。

请求体

当你需要从客户端（比如浏览器）向你的 API 发送数据时，会把它作为请求体发送。

请求体是客户端发送给你的 API 的数据。响应体是你的 API 发送给客户端的数据。

你的 API 几乎总是需要发送响应体。但客户端不一定总是要发送请求体，有时它们只请求某个路径，可能带一些查询参数，但不会发送请求体。

使用 Pydantic 模型来声明请求体，能充分利用它的功能和优点。

发送数据应使用以下之一：POST（最常见）、PUT、DELETE 或 PATCH。

规范中没有定义用 GET 请求发送请求体的行为，但 FastAPI 仍支持这种方式，只用于非常复杂/极端的用例。由于不推荐，在使用 GET 时，Swagger UI 的交互式文档不会显示请求体的文档，而且中间的代理可能也不支持它。

from fastapi import FastAPI
from pydantic import BaseModel


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


app = FastAPI()


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

仅使用这些 Python 类型声明，FastAPI 就可以：

  • 以 JSON 形式读取请求体。
  • （在必要时）把请求体转换为对应的类型。
  • 校验数据。
    • 数据无效时返回清晰的错误信息，并指出错误数据的确切位置和内容。
  • 把接收的数据赋值给参数 item。
    • 因为你把函数中的参数类型声明为 Item，所以还能获得所有属性及其类型的编辑器支持（补全等）。
  • 为模型生成 JSON Schema 定义，如果对项目有意义，还可以在其他地方使用它们。
  • 这些 schema 会成为生成的 OpenAPI Schema 的一部分，并被自动文档的 UIs 使用。

请求体 + 路径参数

可以同时声明路径参数和请求体。FastAPI 能识别与路径参数匹配的函数参数应该从路径中获取，而声明为 Pydantic 模型的函数参数应该从请求体中获取。

请求体 + 路径 + 查询参数

也可以同时声明请求体、路径和查询参数。FastAPI 会分别识别它们，并从正确的位置获取数据。

函数参数按如下规则进行识别：

  • 如果该参数也在路径中声明了，它就是路径参数。
  • 如果该参数是（int、float、str、bool 等）单一类型，它会被当作查询参数。
  • 如果该参数的类型声明为 Pydantic 模型，它会被当作请求体
Annotated  和 Body
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    ...

Annotated 是 Python 标准库 typing 中的一个泛型工具（Python 3.9+ 引入）。它允许你将类型与额外的元数据绑定在一起，而不会影响实际的类型检查。

语法：Annotated[原始类型, 元数据1, 元数据2, ...]

FastAPI 大量使用 Annotated 来给参数附加“参数解析说明”，例如从哪里获取数据（路径、查询、请求体、表单等）、验证规则、依赖项等。这样做的好处是：

  • 类型与元数据共存：类型依然是 Item，但额外告诉 FastAPI “这个 Item 应该从请求体中以嵌套方式提取”。

  • 更好的代码可读性：所有关于该参数的声明都写在同一个地方。

  • 兼容类型检查器：像 mypy、Pyright 会忽略 Annotated 中的元数据，只关注原始类型 Item。

在旧版 FastAPI 中，你可能见过直接写 item: Item = Body(...)。现在官方推荐使用 Annotated 风格。

Body 是 FastAPI 提供的一个函数，用于声明参数应从请求体中获取。

  • 默认行为（embed=False）：
    当请求体参数是一个 Pydantic 模型（如 Item）时，FastAPI 会期望请求体的 整个 JSON 直接就是该模型。
    例如：{"name": "foo", "price": 10.5}

  • embed=True：
    要求请求体必须是一个包含单一字段的对象，该字段名与参数名相同（这里是 item）。
    例如：{"item": {"name": "foo", "price": 10.5}}

为什么需要 embed=True？

  • 当你想在请求体中传递多个独立的模型或模型与其他简单参数并存时，为了避免歧义，可以使用 embed=True 来显式指定参数在请求体中的键名。

  • 在你的例子中，只有一个请求体参数 item，使用 embed=True 是可选的，但它会让 API 要求多一层嵌套。

查询参数与字符串校验

FastAPI 允许为参数声明额外的信息和校验。

from typing import Annotated

from fastapi import FastAPI, Query

app = FastAPI()

# 查询参数 q 的类型为 str | None，这意味着它是 str 类型，但也可以是 None。其默认值确实为 None，所以 FastAPI 会知道它不是必填的。
# async def read_items(q: str | None = None):

# 添加约束：即使 q 是可选的，但只要提供了该参数，其长度不能超过 50 个字符。

@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(max_length=50)] = None):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

这里用的是 Query()，因为这是一个查询参数。稍后还会看到 Path()、Body()、Header() 和 Cookie()，它们也接受与 Query() 相同的参数。

添加更多校验

@app.get("/items/") async def read_items( q: Annotated [str | None, Query(min_length=3, max_length=50)] = None, ):

添加正则表达式

你可以定义一个参数必须匹配的正则表达式 pattern：

@app.get("/items/") async def read_items( q: Annotated [ str | None, Query(min_length=3, max_length=50, pattern="^fixedquery$") ] = None, ):

这个特定的正则表达式通过以下规则检查接收到的参数值：

  • ^：必须以接下来的字符开头，前面没有其他字符。
  • fixedquery：值必须精确等于 fixedquery。
  • $：到此结束，在 fixedquery 之后没有更多字符。
默认值

可以使用 None 以外的默认值。

假设你想要声明查询参数 q 的 min_length 为 3，并且默认值为 "fixedquery"：

async def read_items(q: Annotated [str, Query(min_length=3)] = "fixedquery"):

任何类型的默认值（包括 None）都会让该参数变为可选（非必填）。

必填参数

当我们不需要声明更多校验或元数据时，只需不声明默认值就可以让查询参数 q 成为必填参数。

必填，但可以为 None

可以声明一个参数可以接收 None，但它仍然是必填的。这将强制客户端必须发送一个值，即使该值是 None。为此，可以声明 None 是有效类型，但不声明默认值：

async def read_items(q: Annotated [str | None, Query(min_length=3)]):

查询参数列表 / 多个值

当用 Query 显式地定义查询参数时，还可以声明它接收一个值列表，换句话说，接收多个值。

例如，要声明一个可在 URL 中出现多次的查询参数 q ：

async def read_items(q: Annotated [ list [str] | None, Query() ] = None):

要声明类型为 list 的查询参数（如上例），需要显式地使用 Query，否则它会被解释为请求体。

具有默认值的查询参数列表 / 多个值

可以定义在没有给定值时的默认 list：

async def read_items(q: Annotated [ list [str], Query() ] = ["foo", "bar"]):

使用 list

也可以直接使用 list，而不是 list[str]。在这种情况下 FastAPI 不会检查列表的内容。例如，list[int] 会检查（并记录到文档）列表的内容必须是整数。但仅用 list 不会。

声明更多元数据
from typing import Annotated
from fastapi import FastAPI, Query
app = FastAPI()

@app.get("/items/")
async def read_items(
    q: Annotated[
        str | None,
        Query(
            title="Query string",
            description="Query string for the items to search in the database that have a good match",
            min_length=3,
        ),
    ] = None,
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results
别名参数

# 用 alias 参数声明一个别名，FastAPI 会用该别名在 URL 中查找参数值。

async def read_items(q: Annotated[str | None, Query( alias="item-query" )] = None):

弃用参数

假设你不再喜欢这个参数了。由于还有客户端在使用它，你不得不保留一段时间，但你希望文档清楚地将其展示为已弃用。将参数 deprecated=True 传给 Query。

从 OpenAPI 中排除参数

要把某个查询参数从生成的 OpenAPI 模式中排除（从而也不会出现在自动文档系统中），将 Query 的参数 include_in_schema 设为 False。

自定义校验

有些情况下需要做一些无法通过上述参数完成的自定义校验。在这些情况下，可以使用自定义校验函数，该函数会在正常校验之后应用（例如，在先校验值是 str 之后）。

可在 Annotated 中使用 Pydantic 的 AfterValidator 来实现。（Pydantic 还有 BeforeValidator 等）

以下自定义校验器会检查条目 ID 是否以 isbn-（用于 ISBN 书号）或 imdb-（用于 IMDB 电影 URL 的 ID）开头：

import random
from typing import Annotated

from fastapi import FastAPI
from pydantic import AfterValidator

app = FastAPI()

# 数据存储（模拟数据库）
data = {
    "isbn-9781529046137": "The Hitchhiker's Guide to the Galaxy",
    "imdb-tt0371724": "The Hitchhiker's Guide to the Galaxy",
    "isbn-9781439512982": "Isaac Asimov: The Complete Stories, Vol. 2",
}

# ID 格式验证器
def check_valid_id(id: str):
    if not id.startswith(("isbn-", "imdb-")):
        raise ValueError('Invalid ID format, it must start with "isbn-" or "imdb-"')
    return id

@app.get("/items/")
async def read_items(
    id: Annotated[str | None, AfterValidator(check_valid_id)] = None,
):
    if id:
        item = data.get(id)
    else:
        id, item = random.choice(list(data.items()))
    return {"id": id, "name": item}
字符串与 value.startswith()

字符串的 value.startswith() 可以接收一个元组，它会检查元组中的每个值。

如果需要进行需要与外部组件通信的校验，例如数据库或其他 API，改用 FastAPI 依赖项。这些自定义校验器用于只需检查请求中同一份数据即可完成的事情。

总结

可以为参数声明额外的校验和元数据。

通用的校验和元数据：

  • alias
  • title
  • description
  • deprecated

字符串特有的校验：

  • min_length
  • max_length
  • pattern

也可以使用 AfterValidator 进行自定义校验。

路径参数与数值校验

能够以与查询参数和字符串校验相同的方式使用 Query、Path（以及其他你还没见过的类）声明元数据和字符串校验。

数值校验
  • gt：大于（greater than）
  • ge：大于等于（greater than or equal）
  • lt：小于（less than）
  • le：小于等于（less than or equal）
from typing import Annotated
from fastapi import FastAPI, Path, Query
app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
    q: str,
    size: Annotated[float, Query(gt=0, lt=10.5)],
):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if size:
        results.update({"size": size})
    return results

数值校验同样适用于 float 值。

查询参数模型

对于一组具有相关性的查询参数，可以创建一个 Pydantic 模型来声明它们。便于在多个地方复用模型，并且一次性为所有参数声明验证和元数据。

可以使用 Pydantic 的模型配置来 forbid 任何 extra 字段。

请求体

多个参数

混合使用 Path、Query 和请求体参数。或可通过将默认值设置为 None 将请求体参数声明为可选参数，在这种情况下，将从请求体获取的 item 是可选的。

多个请求体参数：函数中有多个请求体参数（两个 Pydantic 模型参数）时，使用参数名称作为请求体中的键（字段名称）。

请求体中的单一值：与使用 Query 和 Path 为查询参数和路径参数定义额外数据的方式相同，FastAPI 提供了一个同等的 Body。例如，为了扩展先前的模型，除了 item 和 user 之外，还想在同一请求体中具有另一个键 importance。使用 Body 指示 FastAPI 将其作为请求体的另一个键进行处理。

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()

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

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.put("/items/{item_id}")
async def update_item(
    item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results

在这种情况下，FastAPI 将期望像这样的请求体：

{
    "item": {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    },
    "user": {
        "username": "dave",
        "full_name": "Dave Grohl"
    },
    "importance": 5
}

同样的，它将转换数据类型，校验，生成文档等。

多个请求体参数和查询参数：除了请求体参数外，还可以在任何需要的时候声明额外的查询参数。由于默认情况下单一值会被解释为查询参数，因此不必显式地添加 Query。

嵌入单个请求体参数：假设只有一个来自 Pydantic 模型 Item 的请求体参数 item。默认情况下，FastAPI 将直接期望这样的请求体。但是，如果期望一个拥有 item 键并在值中包含模型内容的 JSON，就像在声明额外的请求体参数时所做的那样，则可以使用一个特殊的 Body 参数 embed：

async def update_item(item_id: int, item: Annotated [ Item, Body(embed=True) ]):

字段

与在路径操作函数中使用 Query、Path 、Body 声明校验与元数据的方式一样，可以使用 Pydantic 的 Field 在 Pydantic 模型内部声明校验和元数据。

from typing import Annotated

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = Field(
        default=None, title="The description of the item", max_length=300
    )
    price: float = Field(gt=0, description="The price must be greater than zero")
    tax: float | None = None

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
    results = {"item_id": item_id, "item": item}
    return results

实际上，Query、Path 以及接下来其它对象，会创建公共 Param 类的子类的对象，而 Param 本身是 Pydantic 中 FieldInfo 的子类。

Pydantic 的 Field 返回也是 FieldInfo 的类实例。

Body 直接返回的也是 FieldInfo 的子类的对象。后文还会介绍一些 Body 的子类。

注意，从 fastapi 导入的 Query、Path 等对象实际上都是返回特殊类的函数。

注意，模型属性的类型、默认值及 Field 的代码结构与路径操作函数的参数相同，只不过是用 Field 替换了Path、Query、Body。

嵌套模型

使用 FastAPI，可以定义、校验、记录文档并使用任意深度嵌套的模型（归功于Pydantic）。

List字段、带参数类型的List字段、Set类型、嵌套模型、特殊的类型和校验、带有一组子模型的属性、深度嵌套模型、纯列表请求体。

嵌套模型：定义子模型、将子模型用作类型。

声明请求示例数据
Pydantic 模型中的额外 JSON Schema 数据

可以为一个 Pydantic 模型声明 examples，它们会被添加到生成的 JSON Schema 中。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "name": "Foo",
                    "description": "A very nice Item",
                    "price": 35.4,
                    "tax": 3.2,
                }
            ]
        }
    }

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

这些额外信息会原样添加到该模型输出的 JSON Schema 中，并会在 API 文档中使用。

若使用属性 model_config，接收一个 dict，详见 Pydantic 文档：配置 。

设置 "json_schema_extra"，其值为一个 dict，包含希望出现在生成 JSON Schema 中的任意附加数据，包括 examples。

可用同样的技巧扩展 JSON Schema，添加自定义额外信息。例如，为前端用户界面添加元数据等。

在 Pydantic 模型中使用 Field() 时，你也可以声明额外的 examples。

JSON Schema 中的 examples - OpenAPI

在以下任意场景中使用：

  • Path()
  • Query()
  • Header()
  • Cookie()
  • Body()
  • Form()
  • File()

也可以声明一组 examples，这些带有附加信息的示例将被添加到它们在 OpenAPI 中的 JSON Schema 里。

带有 examples 的 Body

向 Body() 传入 examples，其中包含一个期望的数据示例：

也可以传入多个 examples。

额外的数据类型

常见的数据类型，如:

  • int
  • float
  • str
  • bool

可以使用的其他数据类型:

  • UUID:
    • 一种标准的 "通用唯一标识符" ，在许多数据库和系统中用作ID。
    • 在请求和响应中将以 str 表示。
  • datetime.datetime:
    • 一个 Python datetime.datetime.
    • 在请求和响应中将表示为 ISO 8601 格式的 str ，比如: 2008-09-15T15:53:00+05:00.
  • datetime.date:
    • Python datetime.date.
    • 在请求和响应中将表示为 ISO 8601 格式的 str ，比如: 2008-09-15.
  • datetime.time:
    • 一个 Python datetime.time.
    • 在请求和响应中将表示为 ISO 8601 格式的 str ，比如: 14:23:55.003.
  • datetime.timedelta:
    • 一个 Python datetime.timedelta.
    • 在请求和响应中将表示为 float 代表总秒数。
    • Pydantic 也允许将其表示为 "ISO 8601 时间差异编码", 查看文档了解更多信息。
  • frozenset:
    • 在请求和响应中，作为 set 对待：
      • 在请求中，列表将被读取，消除重复，并将其转换为一个 set。
      • 在响应中 set 将被转换为 list 。
      • 产生的模式将指定那些 set 的值是唯一的 (使用 JSON Schema 的 uniqueItems)。
  • bytes:
    • 标准的 Python bytes。
    • 在请求和响应中被当作 str 处理。
    • 生成的模式将指定这个 str 是 binary "格式"。
  • Decimal:
    • 标准的 Python Decimal。
    • 在请求和响应中被当做 float 一样处理。
  • 可以在这里检查所有有效的 Pydantic 数据类型: Pydantic data types。
更新数据
PUT 替换式更新

HTTP PUT 操作

把输入数据转换为以 JSON 格式存储的数据（比如，使用 NoSQL 数据库时），可以使用 jsonable_encoder。例如，把 datetime 转换为 str。

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str | None = None
    description: str | None = None
    price: float | None = None
    tax: float = 10.5
    tags: list[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

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

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

用 PUT 把数据项 bar 更新为以下请求体时：

{
    "name": "Barz",
    "price": 3,
    "description": None,
}

因为其中未包含已存储的属性 "tax": 20.2，输入模型会取 "tax": 10.5 的默认值。因此，保存的数据会带有这个“新的” tax 值 10.5。

用 PATCH 进行部分更新

使用 HTTP PATCH 操作对数据进行部分更新。即只需发送想要更新的数据，其余数据保持不变。

使用 Pydantic 的 exclude_unset 参数

接收部分更新，在 Pydantic 模型的 .model_dump() 中使用 exclude_unset 参数。

比如，item.model_dump(exclude_unset=True)

这会生成一个 dict，只包含创建 item 模型时显式设置的数据，不包含默认值。再用其生成一个只含已设置（在请求中发送）数据、且省略默认值的 dict。

使用 Pydantic 的 update 参数

接下来，用 .model_copy() 为已有模型创建副本，并传入 update 参数，值为包含更新数据的 dict。

例如，stored_item_model.model_copy(update=update_data)

@app.patch("/items/{item_id}")
async def update_item(item_id: str, item: Item) -> Item:
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.model_dump(exclude_unset=True)
    updated_item = stored_item_model.model_copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)
    return updated_item

应用部分更新应当：

  • （可选）使用 PATCH 而不是 PUT。
  • 提取已存储的数据。
  • 把该数据放入 Pydantic 模型。
  • 生成不含输入模型默认值的 dict（使用 exclude_unset）。
    • 这样只会更新用户实际设置的值，而不会用模型中的默认值覆盖已存储的值。
  • 为已存储的模型创建副本，用接收到的部分更新数据更新其属性（使用 update 参数）。
  • 把模型副本转换为可存入数据库的形式（比如，使用 jsonable_encoder）。
    • 这类似于再次调用模型的 .model_dump() 方法，但会确保（并转换）值为可转换为 JSON 的数据类型，例如把 datetime 转换为 str。
  • 把数据保存至数据库。
  • 返回更新后的模型。

Cookie参数

定义 Cookie 参数与定义 Query 和 Path 参数一样。

Cookie 是什么？

Cookie 是服务器发送到用户浏览器并保存在本地的一小块数据。浏览器在后续请求中会自动将这些数据携带回服务器，用于实现状态管理——因为 HTTP 协议本身是无状态的。

Cookie 的主要用途

  • 会话管理：用户登录状态、购物车内容等；

  • 个性化设置：语言偏好、主题颜色；

  • 跟踪分析：记录用户行为（需注意隐私合规）。

Cookie 的工作流程

  1. 服务器通过响应头 Set-Cookie 告诉浏览器存储一个 Cookie；

  2. 浏览器保存该 Cookie（域名、路径、有效期等属性）；

  3. 之后每次请求同一域名下的资源时，浏览器自动通过请求头 Cookie 将数据发回服务器。

from typing import Annotated

from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
    return {"ads_id": ads_id}

访问 http://localhost:8000/docs (Swagger UI) 或 /redoc：

  • 参数列表：由于使用了 Cookie()，文档会列出三个 Cookie 参数：session_id（必需）、fatebook_tracker（可选）、googall_tracker（可选）。

  • 测试方式：Swagger UI 默认不能直接设置 Cookie，因为浏览器安全限制。但文档会显示参数说明、类型、是否必需以及 extra="forbid" 的约束。

  • 实际测试：可以使用 curl 或浏览器开发者工具手动设置 Cookie 后访问端点；或者在 Swagger UI 中通过 "Try it out" 手动添加 Cookie 头（需修改请求头）。

cURL（Client URL）是一个命令行工具，用于通过 URL 传输数据。它支持几乎所有网络协议（HTTP、HTTPS、FTP 等），最常用于测试 API 接口。

curl.exe "http://localhost:8000/items/" -b "session_id=abc123; fatebook_tracker=xyz"

Header参数

定义 Header 参数的方式与定义 Query、Path、Cookie 参数相同。变量中的下划线，FastAPI 可以自动转换。

类型声明中可以使用 list 定义多个请求头。使用 Python list 可以接收重复请求头所有的值。

from typing import Annotated
from fastapi import FastAPI, Header
app = FastAPI()

@app.get("/items/")
async def read_items(x_token: Annotated[list[str] | None, Header()] = None):
    return {"X-Token values": x_token}

与路径操作通信时，以下面的方式发送两个 HTTP 请求头：

X-Token: foo X-Token: bar

响应结果是：

{ "X-Token values": [ "bar", "foo" ] }

对于一组相关的 header 参数，创建一个 Pydantic 模型来声明。在多个地方能够重用模型，并且可以一次性声明所有参数的验证和元数据。

响应模型-返回类型

可以通过为路径操作函数的返回类型添加注解来声明用于响应的类型。

和为输入数据在函数参数里做类型注解的方式相同，可以使用 Pydantic 模型、list、dict、以及整数、布尔值等标量类型。

response_model参数

在 FastAPI 中，响应模型（response_model 参数）用于声明路径操作函数返回数据的结构。它会自动完成数据过滤、验证、序列化，并影响 OpenAPI 文档生成。

from typing import Any

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = []

# 接收一个 Item 实例（已通过 Pydantic 验证），直接返回它
@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:
    return item

# 返回一个字典列表，每个字典包含 name 和 price，没有 description、tax、tags。
@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:
    return [
        {"name": "Portal Gun", "price": 42.0},
        {"name": "Plumbus", "price": 32.0},
    ]

response_model 的优先级

如果同时声明返回类型和 response_model，response_model 会具有优先级并由 FastAPI 使用。即使返回类型与响应模型不同，也可以为函数添加正确的类型注解，供编辑器和 mypy 等工具使用。同时仍然可以让 FastAPI 使用 response_model 进行数据校验、文档等。

可使用 response_model=None 来禁用该路径操作的响应模型生成；为一些不是有效 Pydantic 字段的东西添加类型注解时，可能需要这样做。

from typing import Any
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
app = FastAPI()

# 创建一个包含明文密码的输入模型和一个不包含它的输出模型
class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: str | None = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

# 将 response_model 声明为不包含密码的 UserOut 模型
@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:
    return user
返回类型与数据过滤

延续上一个例子。希望用一种类型来注解函数，但希望从函数返回的内容实际上可以包含更多数据。 FastAPI 继续使用响应模型来过滤数据。这样即使函数返回了更多数据，响应也只会包含响应模型中声明的字段。在上一个例子中，因为类不同，不得不使用 response_model 参数。但这也意味着无法从编辑器和工具处获得对函数返回类型的检查支持。

不过在大多数需要这样做的场景里，我们只是希望模型像这个例子中那样过滤/移除一部分数据。

在这些场景里，使用类和继承，既利用函数的类型注解获取更好的编辑器和工具支持，又能获得 FastAPI 的数据过滤。

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class BaseUser(BaseModel):
    username: str
    email: EmailStr
    full_name: str | None = None

class UserIn(BaseUser):
    password: str

@app.post("/user/")
async def create_user(user: UserIn) -> BaseUser:
    return user

类型注解与工具链：把函数返回类型注解为 BaseUser，但实际上返回的是一个 UserIn 实例。

编辑器、mypy 和其他工具无异议，因为在类型系统里，UserIn 是 BaseUser 的子类，这意味着当期望 BaseUser 时，返回 UserIn 是合法的。

FastAPI的数据过滤：FastAPI会查看返回类型并确保返回内容只包含该类型中声明的字段。FastAPI 在内部配合 Pydantic 做了多项处理，确保不会把类继承的这些规则用于返回数据的过滤。

其他返回类型注解
  • 直接返回Response。
  • 注解Response的子类。
  • 无效的返回类型注解：当返回其他任意对象（如数据库对象）而它不是有效的 Pydantic 类型，并在函数中按此进行了注解时，FastAPI 会尝试基于该类型注解创建一个 Pydantic 响应模型，但会失败。对于一个在多个类型之间的联合类型，其中一个或多个不是有效的 Pydantic 类型，也会发生同样的情况。
  • 禁用响应模型：设置 response_model=None。
@app.get("/portal", response_model=None)
async def get_portal(teleport: bool = False) -> Response | dict:
    if teleport:
        return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
    return {"message": "Here's your interdimensional portal."}
响应模型的编码参数
使用 response_model_exclude_unset 参数

设置路径操作装饰器参数 response_model_exclude_unset=True，仅返回显式设置的值。响应中将不会包含默认值，而只包含实际设置的值。

还可以使用：

  • response_model_exclude_defaults=True
  • response_model_exclude_none=True
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float = 10.5
    tags: list[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}

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

向该路径操作请求 ID 为 foo 的商品，响应（不包括默认值）将为：

{
    "name": "Foo",
    "price": 50.2
}

如果数据在默认字段中有实际的值，这些值将包含在响应中。比如 ID 为 bar 的项：

{
    "name": "Bar",
    "description": "The bartenders",
    "price": 62,
    "tax": 20.2
}

response_model_include 和 response_model_exclude

它们接收一个由属性名 str 组成的 set，用于包含（忽略其他）或排除（包含其他）这些属性。当只有一个 Pydantic 模型，并且想要从输出中移除一些数据时，可以作为一种快捷方式。

更多模型

多个关联模型这种情况很常见。

特别是用户模型，因为：

  • 输入模型应该含密码
  • 输出模型不应含密码
  • 数据库模型可能需要包含哈希后的密码
Pydantic 的 .model_dump()

返回包含模型数据的 dict。

UserInDB( **user_in.model_dump(), hashed_password = hashed_password )

减少重复

声明 UserBase 模型作为其它模型的基类。然后，用该类衍生出继承其属性（类型声明、校验等）的子类。

Union 或 anyOf

响应可以声明为两个或多个类型的 Union，即该响应可以是这些类型中的任意一种。

在 OpenAPI 中会用 anyOf 表示。使用 Python 标准类型提示 typing.Union 。

  • 定义 Union 类型时，要把更具体的类型写在前面。
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class BaseItem(BaseModel):
    description: str
    type: str

class CarItem(BaseItem):
    type: str = "car"

class PlaneItem(BaseItem):
    type: str = "plane"
    size: int

items = {
    "item1": {"description": "All my friends drive a low rider", "type": "car"},
    "item2": {
        "description": "Music is my aeroplane, it's my aeroplane",
        "type": "plane",
        "size": 5,
    },
}

@app.get("/items/{item_id}", response_model=PlaneItem | CarItem)
async def read_item(item_id: str):
    return items[item_id]
模型列表

同样地，可以声明由对象列表构成的响应。使用标准的 Python list。

@app.get( "/items/", response_model = list [ Item ] )

任意 dict 的响应

使用普通的任意 dict 来声明响应，只需声明键和值的类型，无需使用 Pydantic 模型。如果事先不知道有效的字段/属性名（Pydantic 模型需要预先知道字段）时，这很有用。此时使用 dict。

from fastapi import FastAPI

app = FastAPI()

@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():
    return {"foo": 2.3, "bar": 3.4}

响应状态码

与指定响应模型的方式相同，在以下任意路径操作中，可以使用 status_code 参数声明用于响应的 HTTP 状态码：

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等...

注意，status_code 是（get、post 等）装饰器方法中的参数。与之前的参数和请求体不同，不是路径操作函数的参数。

status_code 参数接收表示 HTTP 状态码的数字。

它可以：

  • 在响应中返回状态码；
  • 在 OpenAPI 概图（及用户界面）中存档。
关于HTTP状态码

在 HTTP 协议中，发送 3 位数的数字状态码是响应的一部分。

这些状态码都具有便于识别的关联名称，但是重要的还是数字。

简言之：

  • 100 - 199 用于返回“信息”。这类状态码很少直接使用。具有这些状态码的响应不能包含响应体。
  • 200 - 299 用于表示“成功”。这些状态码是最常用的：
    • 200 是默认状态码，表示一切“OK”；
    • 201 表示“已创建”，通常在数据库中创建新记录后使用；
    • 204 是一种特殊的例子，表示“无内容”。该响应在没有为客户端返回内容时使用，因此，该响应不能包含响应体。
  • 300 - 399 用于“重定向”。具有这些状态码的响应不一定包含响应体，但 304“未修改”是个例外，该响应不得包含响应体。
  • 400 - 499 用于表示“客户端错误”。这些可能是第二常用的类型：
    • 404，用于“未找到”响应；
    • 对于来自客户端的一般错误，可以只使用 400。
  • 500 - 599 用于表示服务器端错误。几乎永远不会直接使用这些状态码。应用代码或服务器出现问题时，会自动返回这些状态码。

可以使用 fastapi.status 中的快捷变量。

表单数据

当需要接收表单字段而不是 JSON 时，可以使用 Form。Form 是直接继承自 Body 的类。

$ pip install python-multipart

定义 Form 参数

创建表单参数的方式与 Body 或 Query 相同：

from typing import Annotated

from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/")
async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
    return {"username": username}
关于“表单字段”

表单数据通常使用“媒体类型” application/x-www-form-urlencoded 进行编码。

但当表单包含文件时，会编码为 multipart/form-data。

在一个路径操作中可声明多个 Form 参数，但不能同时再声明要接收为 JSON 的 Body 字段，因为此时请求体会使用 application/x-www-form-urlencoded 而不是 application/json 进行编码。

这不是 FastAPI 的限制，而是 HTTP 协议的一部分。

表单的 Pydantic 模型

声明一个 Pydantic 模型，包含希望接收的表单字段，然后将参数声明为 Form。可以在文档 UI 中验证它，地址为 /docs。可以使用 Pydantic 的模型配置来 forbid 任何 extra 字段。

请求文件
定义File参数

使用 File 定义由客户端上传的文件。上传文件是以「表单数据」发送的。File 是直接继承自 Form 的类。注意，从 fastapi 导入的 Query、Path、File 等项，实际上是返回特定类的函数。

如果把路径操作函数参数的类型声明为 bytes，FastAPI 会读取文件，并以 bytes 的形式接收其内容。这意味着整个内容会存储在内存中，适用于小型文件。

在多数情况下，使用 UploadFile 会更有优势。

含 UploadFile 的文件参数

与 bytes 相比，使用 UploadFile 有多项优势：

  • 无需在参数的默认值中使用 File()。
  • 它使用“spooled”文件：
    • 文件会先存储在内存中，直到达到最大上限，超过该上限后会写入磁盘。
  • 因此，非常适合处理图像、视频、大型二进制等大文件，而不会占用所有内存。
  • 你可以获取上传文件的元数据。
  • 它提供 file-like 的 async 接口。
  • 它暴露了一个实际的 Python SpooledTemporaryFile 对象，你可以直接传给期望「file-like」对象的其他库。
UploadFile

UploadFile 的属性如下：

  • filename：上传的原始文件名字符串（str），例如 myimage.jpg。
  • content_type：内容类型（MIME 类型 / 媒体类型）的字符串（str），例如 image/jpeg。
  • file：SpooledTemporaryFile（一个 file-like 对象）。这是实际的 Python 文件对象，可以直接传递给其他期望「file-like」对象的函数或库。

UploadFile 具有以下 async 方法。它们都会在底层调用对应的文件方法（使用内部的 SpooledTemporaryFile）。

  • write(data)：将 data (str 或 bytes) 写入文件。
  • read(size)：读取文件中 size (int) 个字节/字符。
  • seek(offset)：移动到文件中字节位置 offset (int)。
    • 例如，await myfile.seek(0) 会移动到文件开头。
    • 如果先运行过 await myfile.read()，然后需要再次读取内容时，这尤其有用。
  • close()：关闭文件。

由于这些方法都是 async 方法，需要对它们使用 await。

FastAPI 的 UploadFile 直接继承自 Starlette 的 UploadFile，但添加了一些必要的部分，使其与 Pydantic 以及 FastAPI 的其他部分兼容。

什么是「表单数据」

HTML 表单向服务器发送数据的方式通常会对数据使用一种「特殊」的编码，这与 JSON 不同。

FastAPI 会确保从正确的位置读取这些数据，而不是从 JSON 中读取。

可以将 File() 与 UploadFile 一起使用，以设置额外的元数据。

多文件上传

FastAPI 支持同时上传多个文件。它们会被关联到同一个通过「表单数据」发送的「表单字段」。

声明一个由 bytes 或 UploadFile 组成的列表（List），接收的也是含 bytes 或 UploadFile 的列表（list）。

from typing import Annotated

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.post("/files/")
async def create_files(files: Annotated[list[bytes], File()]):
    return {"file_sizes": [len(file) for file in files]}

@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):
    return {"filenames": [file.filename for file in files]}

@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
    """
    return HTMLResponse(content=content)
  • 可以为 File() 设置额外参数，即使是 UploadFile

总结：使用 File、bytes 和 UploadFile 来声明在请求中上传的文件，它们以表单数据发送。

FastAPI 支持同时使用 File 和 Form 定义文件和表单字段。

处理错误

向客户端返回 HTTP 错误响应，可以使用 HTTPException。

HTTPException 是额外包含了和 API 有关数据的常规 Python 异常。

因为是 Python 异常，所以不能 return，只能 raise。

这也意味着，如果在路径操作函数里调用的某个工具函数内部触发了 HTTPException，那么路径操作函数中后续的代码将不会继续执行，请求会立刻终止，并把 HTTPException 的 HTTP 错误发送给客户端。

可以使用与 Starlette 相同的异常处理工具添加自定义异常处理器。

覆盖请求异常验证

请求中包含无效数据时，FastAPI 内部会触发 RequestValidationError。

导入 RequestValidationError，并用 @app.exception_handler(RequestValidationError) 装饰异常处理器。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# 返回一个纯文本响应。客户端收到的错误响应不再是默认的 JSON（如 {"detail":"Not Found"}），而是纯文本字符串，例如 "Nope! I don't like 3."。
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)

# 状态码固定为 400（Bad Request），而不是默认的 422。
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc: RequestValidationError):
    message = "Validation errors:"
    for error in exc.errors():
        message += f"\nField: {error['loc']}, Error: {error['msg']}"
    return PlainTextResponse(message, status_code=400)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

FastAPI 的 HTTPException 错误类继承自 Starlette 的 HTTPException 错误类。

FastAPI 的 HTTPException 在 detail 字段中接受任意可转换为 JSON 的数据，

而 Starlette 的 HTTPException 只接受字符串。

复用 FastAPI 的异常处理器

如果需要在自定义处理后仍复用 FastAPI 的默认异常处理器，可以从 fastapi.exception_handlers 导入并复用这些默认处理器。

路径操作配置

通过传递参数给路径操作装饰器，即可轻松地配置路径操作、添加元数据。

状态响应码
  • 路径操作的响应中定义（HTTP）status_code
  • 直接传递 int 代码，比如404
  • 用 status 的快捷常量
标签

可以通过传入由 str 组成的 list（通常只有一个 str）的参数 tags，为路径操作添加标签。

  • 也可以使用 Enum 的标签维护确保相关的路径操作使用相同的标签。

summary  和 description 添加摘要和描述。

描述内容比较长且占用多行时，可以在函数的 docstring 中声明路径操作的描述，FastAPI 会从中读取。文档字符串支持 Markdown，能正确解析和显示 Markdown 的内容，但要注意文档字符串的缩进。它会在网页交互文档中使用。

response_description 参数用于定义响应的描述说明。

注意，response_description 只用于描述响应，description 一般则用于描述路径操作。

OpenAPI 规定每个路径操作都要有响应描述。

如果没有定义响应描述，FastAPI 则自动生成内容为 "Successful response" 的响应描述。

弃用路径但不删除，可传入 deprecated 参数。

JSON兼容编码器：使用 jsonable_enconder

这个操作不会返回一个包含JSON格式（作为字符串）数据的庞大的str。它将返回一个Python标准数据结构（例如dict），其值和子值都与JSON兼容

依赖项

在编程中，「依赖注入」指的是，代码（本文中为路径操作函数）声明其运行所需并要使用的东西：“依赖”。由该系统（本文中为 FastAPI）负责执行所有必要的逻辑，为代码提供这些所需的依赖（“注入”依赖）。

当你需要以下内容时，这非常有用：

  • 共享业务逻辑（同一段代码逻辑反复复用）
  • 共享数据库连接
  • 实施安全、认证、角色权限等要求
  • 以及更多其他内容...

同时尽量减少代码重复。

在函数参数中使用 Depends 的方式与 Body、Query 等相同，但 Depends 的工作方式略有不同。这里只能给 Depends 传入一个参数。这个参数必须是类似函数的可调用对象。

“依赖注入”的其他常见术语包括：

  • 资源（resources）
  • 提供方（providers）
  • 服务（services）
  • 可注入（injectables）
  • 组件（components）

依赖注入系统的简洁让 FastAPI 能与以下内容兼容：

  • 各类关系型数据库
  • NoSQL 数据库
  • 外部包
  • 外部 API
  • 认证与授权系统
  • API 使用监控系统
  • 响应数据注入系统
  • 等等...
类作为依赖项

依赖项应该是 "可调用对象"。 FastAPI 检查的是它是一个 "可调用对象"（函数，类或其他任何类型）以及定义的参数。

子依赖项

FastAPI 支持创建含子依赖项的依赖项。并可以按需声明任意深度的子依赖项嵌套层级。（能够声明任意嵌套深度的「图」或树状的依赖结构。）FastAPI 负责处理解析不同深度的子依赖项。

多次使用同一个依赖项

如果在同一个路径操作 多次声明了同一个依赖项，例如，多个依赖项共用一个子依赖项，FastAPI 在处理同一请求时，只调用一次该子依赖项。

FastAPI 不会为同一个请求多次调用同一个依赖项，而是把依赖项的返回值进行「缓存」，并把它传递给同一请求中所有需要使用该返回值的「依赖项」。

在高级使用场景中，如果不想使用「缓存」值，而是为需要在同一请求的每一步操作（多次）中都实际调用依赖项，可以把 Depends 的参数 use_cache 的值设置为 False:

async def needy_dependency ( fresh_value : Annotated [ str, Depends ( get_value, use_cache = False ) ] ) :
    return { "fresh_value" : fresh_value }

路径操作装饰器依赖项

当不需要在路径操作函数中使用依赖项的返回值，或有些依赖项不返回值，但仍要执行或解析该依赖项。对于这种情况，不必在声明路径操作函数的参数时使用 Depends，而是可以在路径操作装饰器中添加一个由 dependencies 组成的 list。

from typing import Annotated

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

# 路径装饰器依赖项可以声明请求的需求项（比如响应头）或其他子依赖项/也可以触发异常
async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

# 无论路径装饰器依赖项是否返回值，路径操作都不会使用这些值。

async def verify_key(x_key: Annotated[str, Header()]):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

# 路径操作装饰器支持可选参数 dependencies。该参数的值是由 Depends() 组成的 list：
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]
全局依赖项

当需要为整个应用添加依赖项时，通过与将 dependencies 添加到路径操作装饰器 类似的方式，可以把依赖项添加至整个 FastAPI 应用。

这样一来，就可以为所有路径操作应用该依赖项：

app = FastAPI( dependencies = [ Depends(verify_token), Depends(verify_key) ] )

使用yield的依赖项

FastAPI 支持那些在完成后执行一些额外步骤的依赖项。为此，使用 yield 而不是 return，并把这些额外步骤（代码）写在后面。

  • 确保在每个依赖里只使用一次 yield。

任何可以与以下装饰器一起使用的函数：

都可以作为 FastAPI 的依赖项。实际上，FastAPI 在内部就是用的这两个装饰器。

使用 yield 的数据库依赖项

例如，可以用这种方式创建一个数据库会话，并在完成后将其关闭。

同时使用 yield 和 try 的依赖项

在带有 yield 的依赖中使用了 try 代码块，将收到使用该依赖时抛出的任何异常。

例如，如果在中间的某处代码中（在另一个依赖或在某个路径操作中）发生了数据库事务“回滚”或产生了其他异常，你会在你的依赖中收到这个异常。

因此，可以在该依赖中用 except SomeException 来捕获这个特定异常。同样地，可以使用 finally 来确保退出步骤一定会被执行，无论是否发生异常。

from fastapi import Depends

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()
  • yield 之前的代码：资源初始化
  • yield 返回给路径操作函数使用
  • yield 之后的代码：清理资源（类似 finally）

这非常适合：

  • 数据库连接
  • 文件句柄
  • 网络连接
  • 锁（Lock）
  • 临时资源
使用 yield 的子依赖项

可以声明任意大小和形状的子依赖及其“树”，其中任意一个或全部都可以使用 yield。FastAPI 会确保每个带有 yield 的依赖中的“退出代码”按正确的顺序运行。

async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()

async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)
同时使用 yield 和 HTTPException 的依赖项
同时使用 yield 和 except 的依赖项

如果你在带有 yield 的依赖中使用 except 捕获了一个异常，并且你没有再次抛出它（或抛出一个新异常），FastAPI 将无法察觉发生过异常，就像普通的 Python 代码那样：

from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

class InternalError(Exception):
    pass

# 客户端error响应，但服务器将没有任何日志或其他关于error的提示
def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("Oops, we didn't raise again, Britney 😱")

@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

在这种情况下，客户端会像预期那样看到一个 HTTP 500 Internal Server Error 响应，因为我们没有抛出 HTTPException 或类似异常，但服务器将没有任何日志或其他关于错误是什么的提示。

在带有 yield 和 except 的依赖中务必 raise

如果你在带有 yield 的依赖中捕获到了一个异常，除非你抛出另一个 HTTPException 或类似异常，否则你应该重新抛出原始异常。

可以使用 raise 重新抛出同一个异常：

def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise

现在客户端仍会得到同样的 HTTP 500 Internal Server Error 响应，但服务器日志中会有我们自定义的 InternalError。😎

使用 yield 的依赖项的执行

执行顺序大致如下图所示。时间轴从上到下，每一列都代表交互或执行代码的一部分。

图中主要参与者：

  1. Client（客户端）
  2. Exception Handler（异常处理器）
  3. Dependency with yield（带 yield 的依赖）
  4. Path Operation（路由函数）
  5. Background Tasks（后台任务）

只会向客户端发送一次响应。它可能是某个错误响应，或者是来自 路径操作 的响应。

# 图示表达 yield 依赖项本质上相当于：
resource = create()

try:
    # 路由函数使用 resource
    ...
finally:
    cleanup(resource)

进入依赖→创建资源→yield→执行路由→返回响应/处理异常→执行 yield 后代码→释放资源

提前退出与 scope

通常，带有 yield 的依赖的退出代码会在响应发送给客户端之后执行。但如果在从 路径操作函数 返回之后不再需要使用该依赖，可以使用 Depends(scope="function") 告诉 FastAPI：应当在 路径操作函数 返回后、但在响应发送之前关闭该依赖。

from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()

def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")

@app.get("/users/me")
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]):
    return username

Depends() 接收一个 scope 参数，可为：

  • "function"：在处理请求的 路径操作函数 之前启动依赖，在 路径操作函数 结束后结束依赖，但在响应发送给客户端之前。因此，依赖函数将围绕这个路径操作函数执行。
  • "request"：在处理请求的 路径操作函数 之前启动依赖（与使用 "function" 时类似），但在响应发送给客户端之后结束。因此，依赖函数将围绕这个请求与响应周期执行。

如果未指定且依赖包含 yield，则默认 scope 为 "request"。

子依赖的 scope

当声明一个 scope="request"（默认）的依赖时，任何子依赖也需要有 "request" 的 scope。

但一个 scope 为 "function" 的依赖可以有 scope 为 "function" 和 "request" 的子依赖。

任何依赖都需要能够在子依赖之前运行其退出代码，因为它的退出代码中可能还需要使用这些子依赖。

包含 yield、HTTPException、except 和后台任务的依赖项

带有 yield 的依赖项随着时间演进以涵盖不同的用例并修复了一些问题。如果你想了解在不同 FastAPI 版本中发生了哪些变化，可以在进阶指南中阅读更多：高级依赖项 —— 包含 yield、HTTPException、except 和后台任务的依赖项。

上下文管理器

“上下文管理器”是你可以在 with 语句中使用的任意 Python 对象。

# 比如 with 读取文件。在底层open("./somefile.txt")会创建一个“上下文管理器”对象
with open("./somefile.txt") as f:
    contents = f.read()
    print(contents)

当 with 代码块结束时，它会确保文件被关闭，即使期间发生了异常。

当你用 yield 创建一个依赖时，FastAPI 会在内部为它创建一个上下文管理器，并与其他相关工具结合使用。

在 Python 中，可以通过创建一个带有 __enter__() 和 __exit__() 方法的类来创建上下文管理器。

也可以在 FastAPI 的带有 yield 的依赖中，使用依赖函数内部的 with 或 async with 语句来使用它们：

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()

async def get_db():
    with MySuperContextManager() as db:
        yield db

安全性

OAuth2

OAuth2 是一个授权框架，允许第三方应用在用户授权的前提下，有限制地访问用户在另一服务上存储的资源（如获取个人资料、发帖），而无需获取用户的密码。

  • 核心角色：资源所有者（用户）、客户端（第三方应用）、授权服务器、资源服务器。

  • 常见场景：“使用微信/Google 登录本 App”背后的授权流程。

  • 关键凭证：access_token（访问令牌）代替密码，具有时效性和特定权限范围。

  • 在 FastAPI 中：通过 OAuth2PasswordBearer 等工具轻松实现密码流、Bearer token 验证。

OAuth2 只负责授权，不处理用户身份认证（即不回答“用户是谁”）。

OpenID Connect

OpenID Connect 是基于 OAuth2 的身份认证层，它在 OAuth2 的基础上增加了可验证的用户身份信息，专门解决“登录”问题。

  • 扩展 OAuth2：通过新增 id_token（JWT 格式）来携带用户身份信息（如 sub、name、email）。

  • 核心端点：/userinfo 端点返回用户资料。

  • 常见场景：企业单点登录（SSO）、使用 Google 账户登录第三方网站。

  • 在 FastAPI 中：通常借助 python-jose 解析 id_token，或通过 authlib 等库集成 OIDC 提供商。

OIDC = OAuth2 + 身份认证 + 标准化的用户信息接口。

OpenAPI

OpenAPI 是一种RESTful API 描述规范（原名 Swagger），使用 JSON 或 YAML 格式定义 API 的路径、参数、请求/响应结构、安全方案等。

  • 与安全的关系：在 OpenAPI 文档中通过 securitySchemes 字段声明 API 支持的安全机制（如 OAuth2、API Key、HTTP Bearer）。

  • 在 FastAPI 中：

    • FastAPI 基于 Python 类型注解和 docstring 自动生成 OpenAPI 文档（/openapi.json）。

    • 当你使用 OAuth2PasswordBearer 或 Security() 时，FastAPI 会自动将 OAuth2 流程写入 OpenAPI 文档，Swagger UI 就能展示“Authorize”按钮，方便调试。

  • 作用：OpenAPI 是机器可读的 API 蓝图，支持自动生成客户端 SDK、Mock 服务、文档界面。

OpenAPI 不是安全协议，而是描述 API（包括安全方案）的标准格式。

OpenAPI 定义了以下安全方案：

  • apiKey：一个特定于应用程序的密钥，可以来自：
    • 查询参数。
    • 请求头。
    • cookie。
  • http：标准的 HTTP 身份认证系统，包括：
    • bearer: 一个值为 Bearer 加令牌字符串的 Authorization 请求头。这是从 OAuth2 继承的。
    • HTTP Basic 认证方式。
    • HTTP Digest，等等。
  • oauth2：所有的 OAuth2 处理安全性的方式（称为「流程」）。 *以下几种流程适合构建 OAuth 2.0 身份认证的提供者（例如 Google，Facebook，X (Twitter)，GitHub 等）： * implicit * clientCredentials * authorizationCode
    • 但是有一个特定的「流程」可以完美地用于直接在同一应用程序中处理身份认证：
      • password：接下来的几章将介绍它的示例。
  • openIdConnect：提供了一种定义如何自动发现 OAuth2 身份认证数据的方法。
    • 此自动发现机制是 OpenID Connect 规范中定义的内容。

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

# 创建 OAuth2PasswordBearer 类实例时，需要传入 tokenUrl 参数。
# 该参数包含客户端（运行在用户浏览器中的前端）用来发送 username 和 password 以获取令牌的 URL。
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# oauth2_scheme 变量是 OAuth2PasswordBearer 的一个实例，同时它也是“可调用”的。
@app.get("/items/")
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
    return {"token": token}

这里的 tokenUrl="token" 指向的是尚未创建的相对 URL token，等价于 ./token。

它会在请求中查找 Authorization 请求头，检查其值是否为 Bearer 加上一些令牌，并将该令牌作为 str 返回。如果没有 Authorization 请求头，或者其值不包含 Bearer 令牌，它会直接返回 401 状态码错误（UNAUTHORIZED）。无需检查令牌是否存在即可返回错误；只要你的函数被执行，就可以确定会拿到一个 str 类型的令牌。

获取当前用户
from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
# 创建 Pydantic 用户模型
from pydantic import BaseModel

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

def fake_decode_token(token):
    return User(
        username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
    )

# 创建 get_current_user 依赖项，oauth2_scheme 作为依赖项
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    user = fake_decode_token(token)
    return user

# 注入当前用户：在路径操作 的 Depends 中使用 get_current_user。
# current_user的类型声明为Pydantic的User模型。有助于在函数内部使用代码补全和类型检查。
@app.get("/users/me")
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]):
    return current_user

开发者可以使用任何模型或数据满足安全需求（本例中是 Pydantic 的 User 模型）。

安全和依赖注入的代码只需要写一次。（最后三行代码）

所有端点（路径操作）（或它们的任何部件）都可以使用同一个安全系统，利用这些依赖项或任何其它依赖项。

OAuth2 实现简单的  Password 和 Bearer 验证

OAuth2 规范要求使用“密码流”时，客户端或用户必须以表单数据形式发送 username 和 password 字段。

OAuth2 还支持客户端发送scope表单字段。虽然表单字段的名称是 scope（单数），但实际上，它是以空格分隔的，由多个scope组成的长字符串。

作用域只是不带空格的字符串。

常用于声明指定安全权限，例如：

  • 常见用例为，users:read 或 users:write
  • 脸书和 Instagram 使用 instagram_basic
  • 谷歌使用 https://www.googleapis.com/auth/drive

token 端点的响应必须是 JSON 对象。响应返回的内容应该包含 token_type，以及access_token 字段，它是包含权限 Token 的字符串。

使用密码（及哈希）的OAuth2，基于JWT的Bearer令牌

JWT 意为 “JSON Web Tokens”。它是一种标准，把一个 JSON 对象编码成没有空格、很密集的一长串字符串。

它不是加密的，所以任何人都可以从内容中恢复信息。但它是“签名”的。因此，当你收到一个自己签发的令牌时，你可以验证它确实是你签发的。

pip install pyjwt
pip install "pwdlib[argon2]"

  • 使用 pwdlib，甚至可以把它配置为能够读取由 Django、Flask 安全插件或其他许多工具创建的密码。例如，在数据库中让一个 Django 应用和一个 FastAPI 应用共享同一份数据。或者在使用同一个数据库的前提下，逐步迁移一个 Django 应用到 FastAPI。同时，用户既可以从 Django 应用登录，也可以从 FastAPI 应用登录。

使用下列命令生成一个安全的随机密钥：

$ openss rand -hex 32

把输出复制到变量 SECRET_KEY。

from datetime import datetime, timedelta, timezone
# 使用真正的密码哈希（Argon2）和 JWT 令牌，可用于生产环境（需适当调整密钥）。
from typing import Annotated

import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from pydantic import BaseModel

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "32dbd3b65f26b96889d8384ed4c20452e3a5e9a5cf9e6cc0d7edeeb379097c65"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$argon2id$v=19$m=65536,t=3,p=4$wagCPXjifgvUFBzq4hqe3w$CYaIb8sB+wtD+Vu/P4uod1+Qof8h+1g7bbDlBID48Rc",
        "disabled": False,
    }
}

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

class UserInDB(User):
    hashed_password: str

password_hash = PasswordHash.recommended()

DUMMY_HASH = password_hash.hash("dummypassword")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

def verify_password(plain_password, hashed_password):
    return password_hash.verify(plain_password, hashed_password)

def get_password_hash(password):
    return password_hash.hash(password)

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)

# 验证用户名和密码，使用 DUMMY_HASH 防御时序攻击。
# 当使用一个在数据库中不存在的用户名调用 authenticate_user 时，仍对虚拟哈希运行 verify_password。
# 确保无论用户名是否有效，端点的响应时间大致相同，从而防止可用于枚举已存在用户名的“时间攻击”（timing attacks）。
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        verify_password(password, DUMMY_HASH)
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

# 生成 JWT，设置过期时间
def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# # 依赖项：解码 token，提取 sub，查询用户。若失败则抛出 401。
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

# 依赖项：确保用户未被禁用。
async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# 更新 /token 路径操作
# 用令牌的过期时间创建一个 timedelta。创建一个真正的 JWT 访问令牌并返回它。
@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> Token:
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")

@app.get("/users/me/")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)],
) -> User:
    return current_user

@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return [{"item_id": "Foo", "owner": current_user.username}]

JWT 规范中有一个 sub 键，表示令牌的“主题”（subject）。

sub 键在整个应用中应该是一个唯一标识符，并且它应该是字符串。

OAuth2 支持 “scopes”（作用域）。可以用它们为 JWT 令牌添加一组特定的权限。然后把这个令牌直接交给用户或第三方，在一组限制条件下与 API 交互。

中间件

可以向 FastAPI 应用添加中间件。

“中间件”是一个函数，它会在每个特定的路径操作处理每个请求之前运行，也会在返回每个响应之前运行。

  • 它接收你的应用的每一个请求。
  • 然后它可以对这个请求做一些事情或者执行任何需要的代码。
  • 然后它将这个请求传递给应用程序的其他部分（某个路径操作）处理。
  • 之后它获取应用程序生成的响应（由某个路径操作产生）。
  • 它可以对该响应做一些事情或者执行任何需要的代码。
  • 然后它返回这个响应。

如果有使用 yield 的依赖，依赖中的退出代码会在中间件之后运行。

如果有任何后台任务，它们会在所有中间件之后运行。

在函数的顶部使用装饰器 @app.middleware("http")来创建中间件。

中间件函数会接收：

  • request。
  • 一个函数 call_next，它会把 request 作为参数接收。
    • 这个函数会把 request 传递给相应的路径操作。
    • 然后它返回由相应路径操作生成的 response。
  • 在返回之前，可以进一步修改 response。

可以使用 X- 前缀添加专有自定义请求头。

import time

from fastapi import FastAPI, Request

app = FastAPI()

# 该中间件会在每个请求的响应头中增加 X-Process-Time
# 内容为服务器从接收请求到返回响应所花费的时间（秒）。常用于性能监控或调试。
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

使用 @app.middleware() 装饰器或 app.add_middleware() 方法添加多个中间件时，每个新中间件都会包裹应用，形成一个栈。最后添加的中间件是“最外层”的，最先添加的是“最内层”的。

在请求路径上，最外层的中间件先运行。在响应路径上，它最后运行。

例如：

app.add_middleware(MiddlewareA)
app.add_middleware(MiddlewareB)

这会产生如下执行顺序：

  • 请求：MiddlewareB → MiddlewareA → 路由

  • 响应：路由 → MiddlewareA → MiddlewareB

这种栈式行为确保中间件按可预测且可控的顺序执行。

CORS（跨域资源共享）

Cross-Origin Resource Sharing (CORS) 是浏览器为了保护用户安全而实现的一种机制，它允许一个源的网页访问另一个源的资源。

CORS 或者「跨域资源共享」 指浏览器中运行的前端拥有与后端通信的 JavaScript 代码，而后端处于与前端不同的「源」的情况。源是协议（http，https）、域（myapp.com，localhost，localhost.tiangolo.com）以及端口（80、443、8080）的组合。

由于前后端分离的项目里，前端和后端经常运行在不同的源上（例如前端在 http://localhost:8080，后端在 http://localhost），前端的JavaScript代码向后端发请求时，就会触发CORS检查。

要解决这个问题，你需要在FastAPI应用里配置 CORSMiddleware。

CORSMiddleware配置
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,    # 允许的源列表
    allow_credentials=True,    # 后端支持前端请求携带Cookie或Authorization头等凭证
    allow_methods=["*"],    # 允许跨域请求的HTTP方法列表
    allow_headers=["*"],    # 允许所有请求头
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

支持以下参数：

  • allow_origins: 允许进行跨域请求的源列表。为了让应用安全运行，建议在此明确列出你的前端域名，而不是使用通配符"*"，这能有效避免安全隐患。

  • allow_credentials: 设为True时，表示后端支持前端请求携带Cookie或Authorization头等凭证。特别注意：如果启用此选项，不能将allow_origins设为["*"]，而必须指定明确的源列表，否则浏览器会拒绝请求。

  • allow_methods: 允许跨域请求的HTTP方法列表，默认只允许GET方法。可以使用["*"]允许所有标准方法。

  • allow_headers: 允许跨域请求携带的HTTP请求头列表，默认为空[ ]。可以使用 ["*"] 允许所有的请求头。Accept、Accept-Language、Content-Language 以及 Content-Type 这几个请求头在简单 CORS 请求中总是被允许。如果前端请求携带了自定义头（如Authorization或X-API-Key），必须在这里显式列出，否则预检请求会失败。

  • expose_headers: 指示哪些响应头可以被浏览器中的JavaScript访问到，默认为空。

  • max_age: 设定浏览器可以缓存CORS预检请求结果的最长时间（单位：秒）。设置一个合适的缓存时间（如600秒或更长），能有效减少重复的预检请求，提升性能。

CORS预检请求

配置好CORS中间件后，FastAPI能自动处理一项名为“预检请求”（Preflight Request）的重要工作。当一个跨域请求“非比寻常”时，浏览器会先发送一个 OPTIONS请求来“探路”，询问服务器是否允许接下来的实际操作。

一个请求会被视为“非比寻常”（需要预检）的情况有：

  • 使用了PUT、DELETE等 非GET/POST/HEAD的HTTP方法。

  • 携带了自定义请求头（如Authorization）。

  • 发送的Content-Type不是application/x-www-form-urlencoded、multipart/form-data或text/plain（例如是application/json）。

当浏览器发送预检请求时，FastAPI的CORSMiddleware会自动、正确地响应它，这正是配置它带来的核心价值所在。

SQL（关系型）数据库

FastAPI 并不要求使用 SQL（关系型）数据库。可以使用你想用的任何数据库。

以SQLModel为例：

$ pip install sqlmodel

在 SQLModel 中，任何含有 table=True 属性的模型类都是一个表模型。

任何不含有 table=True 属性的模型类都是数据模型，这些实际上只是 Pydantic 模型（附带一些小的额外功能）。有了 SQLModel，就可以利用继承来在所有情况下避免重复所有字段。

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

# 定义一个基类，包含所有Hero模型都有的字段。
class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)

# 定义一个实际的Hero模型，继承自HeroBase，并添加额外的字段
class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str

# 定义一个只包含公共字段的模型，用于返回给客户端。
class HeroPublic(HeroBase):
    id: int

# 用于验证客户端数据。//对于secret_name接收但不返回
class HeroCreate(HeroBase):
    secret_name: str

# 用于更新Hero模型。
class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None
    
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# 使用 HeroCreate 创建并返回 HeroPublic。即创建一个Hero并返回公共字段。
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

# 读取。使用 response_model=list[HeroPublic] 确保正确地验证和序列化数据
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)  # 排除未设置的字段，只更新设置的字段
    hero_db.sqlmodel_update(hero_data)  # 更新Hero模型
    session.add(hero_db)  # 添加到会话
    session.commit()
    session.refresh(hero_db)  # 刷新会话
    return hero_db

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

更大的应用-多个文件

假设项目文件目录为：

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

假设专门用于处理用户逻辑的文件是位于 /app/routers/users.py 的子模块。你希望将与用户相关的路径操作与其他代码分开，以使其井井有条。但它仍然是同一 FastAPI 应用程序/web API 的一部分（它是同一「Python 包」的一部分）。

可以使用 APIRouter 为该模块创建路径操作。

拆分模块示例：在 users.py 文件中，创建 APIRouter 实例，并定义该模块下的所有路由。

# routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
# 假设我们有统一的认证依赖
from .deps import get_current_user

# 创建 APIRouter 实例
# 1. 设置统一路径前缀 "/users"
# 2. 设置 OpenAPI 文档标签 "users"
# 3. 添加模块级别依赖（例如认证）
router = APIRouter(
    prefix="/users",
    tags=["users"],
    dependencies=[Depends(get_current_user)]  # 该模块下所有路由都需要认证
)

class User(BaseModel):
    name: str
    email: str

@router.get("/me", response_model=User)
async def read_current_user(current_user: User = Depends(get_current_user)):
    """获取当前登录用户信息"""
    return current_user

@router.put("/me", response_model=User)
async def update_current_user(user: User, current_user: User = Depends(get_current_user)):
    """更新当前登录用户信息（示例实现）"""
    # 这里可以包含更新数据库的逻辑
    return user

主应用集成：在 main.py 中，像“搭积木”一样，使用 app.include_router() 方法将各个模块的路由挂载到主应用上。

# main.py
from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])

# 将 routers 子模块中的 router 对象引入并挂载
app.include_router(users.router)
app.include_router(items.router)
# 原始的APIRouter不变，但设置了一个自定义的prefix以便所有路径操作以/admin开头
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)

@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}
APIRouter

一个用于将路径操作（路由）组织成独立模块的工具。其核心价值就是将代码进行模块化拆分，每个功能模块（如用户、商品、订单）的路由、依赖项和中间件都可以独立维护，最后再“组装”到主应用中。

APIRouter 支持非常丰富的参数，让你可以精细地控制路由组的行为。

参数 类型 描述
prefix str 路径前缀。为该路由组下的所有路径操作统一添加一个前缀，如 "/users"。
tags List[str or Enum] 文档标签。用于在 OpenAPI 文档中对这些路径操作进行分组，通常设为模块名，让 API 文档更清晰。
dependencies Sequence[Depends] 模块级依赖。声明该路由组下所有端点都需执行的依赖项，常用于统一认证、权限校验等。注意：其返回值默认不会注入到路径操作函数中。
responses Dict 额外响应描述。在 OpenAPI 文档中为路径组添加额外的响应信息，如状态码和描述。
deprecated bool 标记废弃。若设为 True，则会在 API 文档中将该路由组下的所有端点标记为“已废弃”。
include_in_schema bool 隐藏接口。若设为 False，则该路由组下的所有路径操作将不会出现在 OpenAPI 文档中。
... ... 此外，还有 callbacks、default_response_class等高级参数，在特定场景下会非常有用。
APIRouter 进阶技巧

1. 模块化的依赖项与中间件

可以在创建 APIRouter 时通过 dependencies 参数设置模块级依赖，例如为所有需要认证的接口统一添加 Depends(get_current_user)。

对于模块级中间件，虽然不能直接通过 APIRouter 的参数设置，但可以通过创建一个自定义的 APIRoute 子类并将其传入 route_class 参数来实现，这是一种更高级的用法。

2. 嵌套路由

APIRouter 还支持嵌套，即一个路由可以被包含到另一个路由中，最终再被包含到主应用中。这种方式非常适合构建复杂的、分层的 API 结构。

# 内层路由
internal_router = APIRouter()
@internal_router.get("/internal")
def internal_endpoint():
    return {"msg": "This is internal"}

# 外层路由
outer_router = APIRouter(prefix="/outer")
outer_router.include_router(internal_router, prefix="/inner")

# 主应用包含外层路由
app.include_router(outer_router)
# 最终可访问路径为: /outer/inner/internal

3. 插件式架构

利用 APIRouter，可以构建插件式的应用架构。在项目启动时，动态扫描特定目录并自动加载其中的路由模块。这为构建可扩展的应用程序（如后台管理系统、CMS）提供了强大的支持。

在 pyproject.toml 中配置 entrypoint

由于 FastAPI app 对象位于 app/main.py 中，你可以在 pyproject.toml 中这样配置 entrypoint：

[tool.fastapi] entrypoint = "app.main:app"

等价于像这样导入：

from app.main import app

这样 fastapi 命令就知道到哪里去找到你的应用了。

流式传输 JSON Lines

“流式传输”数据意味着你的应用会在整段数据全部准备好之前，就开始把每个数据项发送给客户端。也就是说，它会先发送第一个数据项，客户端会接收并开始处理它，而此时你的应用可能还在生成下一个数据项。

 “JSON Lines”，一种每行发送一个 JSON 对象的格式。响应的内容类型是 application/jsonl（而不是 application/json），响应体类似于：

{"name": "Plumbus", "description": "A multi-purpose household device."}
{"name": "Portal Gun", "description": "A portal opening device."}
{"name": "Meeseeks Box", "description": "A box that summons a Meeseeks."}

它与 JSON 数组（相当于 Python 的 list）非常相似，但不是用 [] 包裹、并在各项之间使用 , 分隔，而是每行一个 JSON 对象，彼此以换行符分隔。

使用 FastAPI 流式传输 JSON Lines

要在 FastAPI 中流式传输 JSON Lines，可以在路径操作函数中不用 return，而是用 yield 逐个产生每个数据项。

如果你要返回的每个 JSON 项是类型 Item（一个 Pydantic 模型），并且这是一个异步函数，可以将返回类型声明为 AsyncIterable[Item]：

@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item

对于非异步的路径操作函数，将返回类型声明为 Iterable[Item] 。

省略返回类型时 FastAPI 会使用 jsonable_encoder 将数据转换为可序列化为 JSON 的形式，然后以 JSON Lines 发送。

服务器发送事件 SSE

可以使用服务器发送事件（SSE）向客户端流式发送数据。这类似于流式传输 JSON Lines，但使用 text/event-stream 格式，浏览器原生通过 EventSource API 支持。

SSE 是一种通过 HTTP 从服务器向客户端流式传输数据的标准。每个事件是一个带有 data、event、id 和 retry 等“字段”的小文本块，以空行分隔。类似于：

data: {"name": "Portal Gun", "price": 999.99}

data: {"name": "Plumbus", "price": 32.99}

SSE 常用于 AI 聊天流式输出、实时通知、日志与可观测性，以及其他服务器向客户端推送更新的场景。

EventSourceResponse

要在 FastAPI 中流式传输 SSE，在路径操作函数中使用 yield，并设置 response_class=EventSourceResponse。

from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse   # 导入
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]

# 每个被 yield 的项会被编码为 JSON，并放入 SSE 事件的 data: 字段发送。
@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item

@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item

@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item

@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse)
def sse_items_no_async_no_annotation():
    for item in items:
        yield item

非 async 的路径操作函数（常规 def 函数）以同样方式使用yield，返回类型为 Iterable[Item] 。

省略返回类型，FastAPI 将使用 jsonable_encoder 转换数据并发送。

ServerSentEvent

如果需要设置 event、id、retry 或 comment 等 SSE 字段，可以 yield ServerSentEvent 对象，而不是直接返回数据。从 fastapi.sse 导入 ServerSentEvent：

from fastapi.sse import EventSourceResponse, ServerSentEvent

@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(comment="stream of item updates")
    for i, item in enumerate(items):
        yield ServerSentEvent(data=item, event="item_update", id=str(i + 1), retry=5000)

data 字段始终会被编码为 JSON。可以传入任何可被序列化为 JSON 的值，包括 Pydantic 模型。

原始数据

如果需要发送不进行 JSON 编码的数据，使用 raw_data 而不是 data。这对于发送预格式化文本、日志行或特殊的 "哨兵" 值（例如 [DONE]）很有用。

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()

#  data 和 raw_data 是互斥的。每个 ServerSentEvent 上只能设置其中一个。
@app.get("/logs/stream", response_class=EventSourceResponse)
async def stream_logs() -> AsyncIterable[ServerSentEvent]:
    logs = [
        "2025-01-01 INFO  Application started",
        "2025-01-01 DEBUG Connected to database",
        "2025-01-01 WARN  High memory usage detected",
    ]
    for log_line in logs:
        yield ServerSentEvent(raw_data=log_line)
使用 Last-Event-ID 恢复

当连接中断后浏览器重新连接时，会在 Last-Event-ID 头中发送上次收到的 id。可以将其读取为一个请求头参数，并据此从客户端离开的地方恢复流：

@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items(
    # 从请求头 Last-Event-ID 中读取客户端最后收到的事件 ID（整型，可选）。
    # 浏览器在重连时会自动发送该头，用于断点续传。
    last_event_id: Annotated[int | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    # 计算起始索引
    start = last_event_id + 1 if last_event_id is not None else 0
    for i, item in enumerate(items):
        if i < start:
            continue
        yield ServerSentEvent(data=item, id=str(i))
使用 POST 的 SSE

SSE 适用于任意 HTTP 方法，不仅仅是 GET。这对像 MCP 这样通过 POST 传输 SSE 的协议很有用：

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()

class Prompt(BaseModel):
    text: str

@app.post("/chat/stream", response_class=EventSourceResponse)
async def stream_chat(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    words = prompt.text.split()
    for word in words:
        yield ServerSentEvent(data=word, event="token")
    # 注意：raw_data="[DONE]" 不会额外添加引号，直接发送 [DONE] 文本
    yield ServerSentEvent(raw_data="[DONE]", event="done")
技术细节
  • 当 15 秒内没有任何消息时，发送一个保活 ping 注释，以防某些代理关闭连接，正如 HTML 规范：Server-Sent Events 中建议的那样。
  • 设置 Cache-Control: no-cache 响应头，防止缓存流。
  • 设置特殊响应头 X-Accel-Buffering: no，以防止某些代理（如 Nginx）缓冲。

后台任务

可以定义在返回响应后运行的后台任务。

这对需要在请求之后执行的操作很有用，但客户端不必在接收响应之前等待操作完成。

包括这些例子：

  • 执行操作后发送的电子邮件通知：
    • 由于连接到电子邮件服务器并发送电子邮件往往很“慢”（几秒钟），您可以立即返回响应并在后台发送电子邮件通知。
  • 处理数据：
    • 例如，假设您收到的文件必须经过一个缓慢的过程，您可以返回一个"Accepted"(HTTP 202)响应并在后台处理它。
BackgroundTasks
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

# 创建要作为后台任务运行的函数
def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    # 将write_notification 添加到后台任务队列中，会在当前请求的响应发送之后异步执行，不会阻塞请求。
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

BackgroundTasks 也适用于依赖注入系统，可以在多个级别声明 BackgroundTasks 类型的参数：在 路径操作函数 里，在依赖中(可依赖)，在子依赖中，等等。

  • 如果需要执行繁重的后台计算，并且不一定需要由同一进程运行（例如，不需要共享内存、变量等），那么使用其他更大的工具（如 Celery）可能更好。它们往往需要更复杂的配置，即消息/作业队列管理器，如RabbitMQ或Redis，但它们允许在多个进程中运行后台任务，甚至是在多个服务器中。
  • 但是，如果需要从同一个FastAPI应用程序访问变量和对象，或者需要执行小型后台任务（如发送电子邮件通知），只需使用 BackgroundTasks 即可。

元数据和文档URL

可以在 FastAPI 应用程序中自定义多个元数据配置。

API 元数据

app = FastAPI(
    title="ChimichangApp",
    description=description,
    summary="Deadpool's favorite app. Nuff said.",
    version="0.0.1",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "dp@x-force.example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)

@app.get("/items/")
async def read_items():
    return [{"name": "Katana"}]

标签元数据

可以通过参数 openapi_tags 为用于分组路径操作的不同标签添加额外的元数据。

它接收一个列表，列表中每个标签对应一个字典。

每个字典可以包含：

  • name（必填）：一个 str，与在路径操作和 APIRouter 的 tags 参数中使用的标签名相同。
  • description：一个 str，该标签的简短描述。可以使用 Markdown，并会显示在文档 UI 中。
  • externalDocs：一个 dict，描述外部文档，包含：
    • description：一个 str，该外部文档的简短描述。
    • url（必填）：一个 str，该外部文档的 URL。

openapi_tags 是 FastAPI 类的一个参数，它接收一个字典列表，每个字典对应一个标签（tag）的元数据配置。这些元数据会写入生成的 OpenAPI 规范（/openapi.json）中，进而在 Swagger UI（/docs）或 ReDoc（/redoc）中展示，用于分组和说明 API 端点。

from fastapi import FastAPI

# 创建标签元数据。注：不必为使用的所有标签都添加元数据。
tags_metadata = [
    {
        "name": "users",
        "description": "Operations with users. The **login** logic is also here.",
    },
    {
        "name": "items",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/",
        },
    },
]

# 传递给 openai_tags 参数
app = FastAPI(openapi_tags=tags_metadata)

@app.get("/users/", tags=["users"])
async def get_users():
    return [{"name": "Harry"}, {"name": "Ron"}]

@app.get("/items/", tags=["items"])
async def get_items():
    return [{"name": "wand"}, {"name": "flying broom"}]

每个标签元数据字典的顺序定义了在文档用户界面显示的顺序。

默认情况下，OpenAPI 模式服务于 /openapi.json。但可以通过参数 openapi_url 对其进行配置。例如，将其设置为服务于 /api/v1/openapi.json：

app = FastAPI ( openapi_url=" /api/v1/openapi.json " )

禁用 OpenAPI 模式，可以将其设置为 openapi_url=None，这样也会禁用使用它的文档用户界面。

可以配置两个文档用户界面，包括：

  • Swagger UI：服务于 /docs。
    • 可以使用参数 docs_url 设置它的 URL。
    • 可以通过设置 docs_url=None 禁用它。
  • ReDoc：服务于 /redoc。
    • 可以使用参数 redoc_url 设置它的 URL。
    • 可以通过设置 redoc_url=None 禁用它。

例如，设置 Swagger UI 服务于 /documentation 并禁用 ReDoc：

app = FastAPI ( docs_url = "/documentation", redoc_url=None )

静态文件

使用 StaticFiles 从目录中自动提供静态文件。

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 将一个 StaticFiles() 实例“挂载”（Mount）到指定路径。
app.mount("/static", StaticFiles(directory="static"), name="static")

“挂载”表示在特定路径添加一个完全“独立”的应用，然后负责处理所有子路径。

  • 第一个 "/static" 指的是这个“子应用”将被“挂载”到的子路径。因此，任何以 "/static" 开头的路径都会由它处理。
  • directory="static" 指的是包含你的静态文件的目录名称。
  • name="static" 为它提供了一个可被 FastAPI 内部使用的名称。
  • 这些参数都可以不是“static”，根据应用需求和具体细节进行调整。

测试

使用 TestClient

$ pip install httpx

from fastapi import FastAPI
from fastapi.testclient import TestClient    # 导入

app = FastAPI()

@app.get("/")
async def read_main():    # test_开头的函数（标准 pytest 约定）
    return {"msg": "Hello World"}

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200    # assert 语句（标准）
    assert response.json() == {"msg": "Hello World"}

分离测试

# test_main.py ，与main.py在同目录下
from .main import app

client = TestClient(app)

调试

导入 uvicorn 并运行。

import uvicorn    # 导入
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}

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

# fastapi
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区
cover

AtomCode 全新上线万亿代码模型 LongCat-2.0。限时 7 天免费体验！

avatar AtomGit开源社区