FastAPI OpenAPI Callbacks 实战：用 `callbacks` 参数把“你的 API 将要回调的外部 API“文档化

FastAPI OpenAPI Callbacks 实战：用callbacks参数把"你的 API 将要回调的外部 API"文档化

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

这篇技术指南以 FastAPI 官方教程的 OpenAPI Callbacks 一章为主线，讲清"回调（callback）"在 API 设计中的含义、为什么需要把回调写进 OpenAPI 文档，以及如何仅用装饰器参数callbacks=就把外部开发者需要实现的外部 API 完整描述出来。读完你将掌握：回调文档代码的组织方式（APIRouter+ 仅含pass的path operation）、OpenAPI 3 Key Expression（如{$callback_url}、{$request.body.id}）的取值规则，以及这些写法如何最终体现在/docs的 Swagger UI 与/openapi.json的callbacks字段中。文中所指代码与截图均来自本仓库 docs_src/openapi_callbacks/tutorial001_py310.py。

什么是 OpenAPI Callback（回调）

你可以构建这样一个 API：它的某个path operation会在运行过程中，主动向"别人（很可能是使用你 API 的那位外部开发者）所创建的 external API"发起一次请求。

当你的 API app 调用那个external API时，这个过程就被称作callback（回调）：外部开发者编写的软件先向你的 API 发来请求，随后你的 API "call back"——反向向某个external API（通常正是同一位开发者写的）再发一个请求。

在这种场景下，你非常有必要文档化"那个 external API 应该长成什么样"：它应该具备什么样的path operation、期望接收什么 body、应该返回什么 response 等等。这正是 FastAPI 的 OpenAPI Callbacks 特性要解决的问题。

场景示例：一个创建发票（Invoice）的 app

教程用"发票应用"串起全部概念。设想你开发了一个允许创建发票的 app，每张发票包含id、title（可选）、customer与total。

你的 API 使用者（一位外部开发者）会通过 POST 请求在你的 API 中创建一张发票。接着，你的 API（假想流程）会：

• 把发票发送给外部开发者的某个客户；
• 完成收款；
• 向 API 使用者（外部开发者）回发一条通知。
  • 这一步通过你的 API向外部开发者提供的一个external API发送 POST 请求来完成——这就是"回调"。

问题在于：回调真正发生的位置在你的服务器上、你的业务代码里，但需要实现那个回调接收端（external API）的却是外部开发者。如果没有契约文档，两边很容易在字段、路径、响应格式上对不上。FastAPI 给出的解法是：用你早已熟悉的path operation写法，把回调端"应该长什么样"声明出来，并交给 Swagger UI 展示。

先看一个普通的 FastAPI app（回调之前的样子）

在加入 callback 之前，一个常规的 app 会有一个接收Invoicebody 的path operation，外加一个携带回调 URL 的 query 参数callback_url：

from fastapi import APIRouter, FastAPI from pydantic import BaseModel, HttpUrl app = FastAPI() class Invoice(BaseModel): id: str title: str | None = None customer: str total: float class InvoiceEvent(BaseModel): description: str paid: bool class InvoiceEventReceived(BaseModel): ok: bool invoices_callback_router = APIRouter() @invoices_callback_router.post( "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass @app.post("/invoices/", callbacks=invoices_callback_router.routes) def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None): """ Create an invoice. This will (let's imagine) let the API user (some external developer) create an invoice. And this path operation will: * Send the invoice to the client. * Collect the money from the client. * Send a notification back to the API user (the external developer), as a callback. * At this point is that the API will somehow send a POST request to the external API with the notification of the invoice event (e.g. "payment successful"). """ # Send the invoice, collect the money, send the notification (the callback) return {"msg": "Invoice received"}

（完整源码见 tutorial001_py310.py。）

这部分代码非常常规，绝大部分写法你应该都很熟悉。其中两点值得单独指出：

1. callback_url使用了 Pydantic 的HttpUrl类型（对应pydantic的 URL 校验网络类型）。这意味着请求/invoices/时若传了callback_url，FastAPI 会自动完成 URL 格式校验，非法的 URL 会直接返回 422 校验错误——主请求的入参就能得到约束。
2. 这里唯一的新东西，是path operation decorator参数里的callbacks=invoices_callback_router.routes。它不参与任何运行时行为，只用于生成文档。下一节说明它到底是什么。

文档化 callback：真正重要的是"契约"

实际发出回调的代码完全取决于你自己的 API app 业务，且不同 app 之间差异极大。它可能只是寥寥一两行：

callback_url = "https://example.com/api/v1/invoices/events/" httpx.post(callback_url, json={"description": "Invoice paid", "paid": True})

回调本质上就是一次 HTTP 请求；自己实现回调时，可用 HTTPX、Requests 之类的客户端库发出去即可。

但 callback 中最重要的部分是：确保你的 API 使用者（外部开发者）能按照你将要发送的数据格式，正确实现那个 external API。因此，教程接下来做的不是实现回调本身（那可能只是一行代码），而是添加代码来文档化"external API 应该如何接收来自你 API 的回调"。

这段文档化代码会出现在你 API 的/docs（Swagger UI）中，让外部开发者一目了然地知道 external API 该怎么搭。

编写 callback 文档代码

需要特别澄清：这些代码永远不会在你的 app 中执行，我们只需要它来"文档化"external API 的样子。好消息是你已经会用 FastAPI 为 API 生成自动文档——现在把同样的知识反向用在"external API 该长什么样"上即可：创建出 external API 应当实现的那些path operation（也就是你的 API 将要调用的那些）。

一个非常实用的写作技巧：编写回调文档代码时，把自己代入"那位外部开发者"的视角——仿佛此刻你在实现 external API，而不是你自己的 API。临时采用这个视角，会让你更自然地判断：参数放在哪里、body 的 Pydantic model 是什么、response 的 model 是什么。

创建承载回调的APIRouter

首先新建一个APIRouter，用来容纳一个或多个回调定义：

from fastapi import APIRouter, FastAPI invoices_callback_router = APIRouter()

编写回调path operation

用同一个APIRouter定义回调path operation。它看起来和普通 FastAPIpath operation几乎一样：

• 声明它将要接收的 body，例如body: InvoiceEvent；
• 声明它应当返回的 response，例如response_model=InvoiceEventReceived。

对应到代码里就是：

class InvoiceEvent(BaseModel): description: str paid: bool class InvoiceEventReceived(BaseModel): ok: bool invoices_callback_router = APIRouter() @invoices_callback_router.post( "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass

与普通path operation相比，它有两个关键差异：

1. 函数体不需要任何真实逻辑——你的 app 永远不会调用这段代码，它只用于生成文档，所以函数体可以只有pass。从源码结构看，这一约束与docs_src及测试中pass # pragma: nocover的写法一致（见 test_sub_callbacks.py），明确表达"仅供文档、不计入覆盖率"。
2. path 中可以包含 OpenAPI 3 Key Expression（表达式），用变量引用"发送到你 API 的原始请求"中的参数与组成部分。这正是回调 URL 能做到"动态拼接"的原理。

理解 callback path expression

本例中回调路径是一个含表达式的字符串：

"{$callback_url}/invoices/{$request.body.id}"

• {$callback_url}：引用原始请求里名为callback_url的参数值（这里是 query 参数）；
• {$request.body.id}：引用原始请求 JSON body 中id字段的值。

来完整推演一遍取值过程。假如外部开发者向你的 API 发出请求：

https://yourapi.com/invoices/?callback_url=https://www.external.org/events

携带如下 JSON body：

{ "id": "2expen51ve", "customer": "Mr. Richie Rich", "total": "9999" }

那么你的 API 处理完发票后，会在稍后的某个时点向callback_url（即 external API）发起这样的回调请求：

https://www.external.org/events/invoices/2expen51ve

请求体大致形如：

{ "description": "Payment celebration", "paid": true }

并期望 external API 返回类似下面的 JSON 响应体：

{ "ok": true }

请注意：最终使用的回调 URL 同时包含了callback_urlquery 参数收到的 URL（https://www.external.org/events），以及来自 JSON body 内部的发票id（2expen51ve）。{$callback_url}/invoices/{$request.body.id}这个模式把两者拼接成了https://www.external.org/events/invoices/2expen51ve。

把回调挂到主path operation

现在你的 callback router 中已经有了所需的回调path operation（即 external developer 需要在 external API 里实现的操作）。接下来，在你的 API 主path operation的 decorator 中通过callbacks参数传入该 router 的.routes属性：

@app.post("/invoices/", callbacks=invoices_callback_router.routes) def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None): ...

需要特别注意：传入的不是 router 本身（invoices_callback_router），而是它的.routes，即invoices_callback_router.routes。FastAPI 会遍历这些 route，用于生成回调的 OpenAPI 文档。

在/docs中查看效果

启动 app 并访问http://127.0.0.1:8000/docs，在POST /invoices/的文档里会多出一个Callbacks区域，直观展示 external API 应当如何实现：

如上图所示，Swagger UI 会渲染出回调名invoice_notification、路径模板{$callback_url}/invoices/{$request.body.id}、必需的 JSON 请求体以及默认的成功响应——外部开发者照此即可实现正确的回调端点。

底层实现：callbacks如何进入 OpenAPI 文档

文档层面的行为背后，是 FastAPI 运行时对callbacks的统一处理，可以沿源码走一遍调用链。

• 接收阶段：在 fastapi/routing.py 中，APIRoute的构造与_populate_api_route_state()都会接收callbacks: list[BaseRoute] | None，并把它直接赋给路由对象（route.callbacks = callbacks）。也就是说，callbacks只是挂在 route 上的"元数据"，不影响请求分发。
• 生成阶段：在 fastapi/openapi/utils.py 的 OpenAPI 生成逻辑中，只要route.callbacks非空，就会对每个APIRoute类型回调递归调用get_openapi_path()（回调本身也是一个完整的path operation，照常生成 parameters、requestBody、responses），再以callbacks[callback.name] = {callback.path: cb_path}的形式写入operation["callbacks"]。这解释了为什么回调的 key 是函数名（如invoice_notification）、value 的 key 是带表达式的路径字符串。
• Schema 收集阶段：同一个文件里，生成components前会把回调声明中用到的模型一并纳入——callback_flat_models.extend(get_fields_from_routes(api_route.callbacks))。因此InvoiceEvent、InvoiceEventReceived这些仅在回调文档中出现的 Pydantic 模型，也会被注册为 OpenAPI schema，Swagger UI 才能正确渲染请求体与响应体。
• 测试验证：仓库中的 test_sub_callbacks.py 用TestClient请求/openapi.json并断言了完整的 schema 快照，可以看到post /invoices/的 operation 下确实包含callbacks字段，其中invoice_notification键、路径表达式、requestBody 对InvoiceEvent的$ref、responses 对InvoiceEventReceived的$ref都与文档描述一一对应。

扩展用法与实战建议

• callbacks不仅可用于单个path operation。从 fastapi/routing.py 的类型与注释可以看出，APIRouter以及include_router(..., callbacks=...)同样接受回调列表，用于"该 router 内所有 path operation 共同生效"的回调。测试 test_sub_callbacks.py 演示了这种用法：在子 router 上按路径操作传入一套回调、再通过include_router(..., callbacks=events_callback_router.routes)附加另一套，最终/openapi.json中两个回调（invoice_notification与event_callback）都被合并进同一个 operation 的callbacks字段。
• 回调文档与回调实现解耦：callbacks=只影响 OpenAPI/Swagger 文档，绝不注册为可被访问的真实路由；真实回调仍需你在业务代码里（如create_invoice函数体中）用 HTTP 客户端主动发出。文中示例的真实实现可能只是一行httpx.post(...)，请勿把"文档里能看见回调"误当作"回调已经会被自动执行"。
• 表达式字段要与实际请求对齐：{$callback_url}必须对应真实 query 参数名，{$request.body.id}必须对应真实 body 中的 JSON 字段；若你的原始请求结构不同，请相应替换表达式中的变量名，并在回调端实现时保持 URL 模板与发送逻辑一致。

综上，FastAPI 的 OpenAPI Callbacks 让你可以在自己的 API 文档里"反向"描述出你将要访问的外部 API，把回调契约从口头约定变成可交互、可校验、自动渲染的 OpenAPI 规范。搭配本仓库的 教程源码、OpenAPI 生成实现 与 schema 快照测试 一同阅读，可以完整掌握从 API 定义到 OpenAPI 输出的全部机制。

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi