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_urlfrom 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_modelInvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass app.post(/invoices/, callbacksinvoices_callback_router.routes) def create_invoice(invoice: Invoice, callback_url: HttpUrl | None None): Create an invoice. This will (lets 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。这部分代码非常常规绝大部分写法你应该都很熟悉。其中两点值得单独指出callback_url使用了 Pydantic 的HttpUrl类型对应pydantic的 URL 校验网络类型。这意味着请求/invoices/时若传了callback_urlFastAPI 会自动完成 URL 格式校验非法的 URL 会直接返回 422 校验错误——主请求的入参就能得到约束。这里唯一的新东西是path operation decorator参数里的callbacksinvoices_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 的/docsSwagger 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_modelInvoiceEventReceived。对应到代码里就是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_modelInvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass与普通path operation相比它有两个关键差异函数体不需要任何真实逻辑——你的 app 永远不会调用这段代码它只用于生成文档所以函数体可以只有pass。从源码结构看这一约束与docs_src及测试中pass # pragma: nocover的写法一致见 test_sub_callbacks.py明确表达仅供文档、不计入覆盖率。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_urlhttps://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 参数收到的 URLhttps://www.external.org/events以及来自 JSON body 内部的发票id2expen51ve。{$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/, callbacksinvoices_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 schemaSwagger 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(..., callbacksevents_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考