FastAPI OpenAPI Callbacks 实战用 Swagger UI 文档化你的 API 回调接口【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi导读OpenAPI Callbacks 是 FastAPI 提供的一种「文档化外部回调接口」的机制当你的 API 会主动向外部开发者提供的另一个 API 发送请求例如支付完成后通知外部系统时你可以直接在 OpenAPI / Swagger UI 中声明这个外部 API 应当长什么样——它应该暴露什么路径、接收什么请求体、返回什么响应。读完本文你将掌握如何在 FastAPI 中定义 callback router、使用 OpenAPI 3 Key Expression如{$callback_url}、{$request.body.id}动态表达回调路径并理解callbacks参数在源码层面的处理链路让你的 API 文档对调用方开发者友好到「照着文档就能实现回调」。一、什么是 OpenAPI Callbacks你可以创建这样一类 API其中的某个路径操作path operation会向某个由他人很可能就是使用你 API 的那位开发者编写的外部 API发起请求。当你的 API 应用调用这个外部 API时整个过程被称为callback回调。原因在于外部开发者编写的软件先向你的 API 发送请求随后你的 API 再回拨calls back向一个外部 API很可能仍由同一位开发者创建发送请求。在这种场景下你可能会希望文档化这个外部 API 应当长什么样它应该具备哪些路径操作、期望接收什么请求体、应当返回什么响应等等。这正是 OpenAPI Callbacks 要解决的问题——它把你的 API 将要调用的外部契约写进 OpenAPI 规范让调用方开发者直接在文档中看到自己需要实现的接口形态。这一功能在 FastAPI 文档体系中与 OpenAPI Webhooks 同属高级用法章节对应源码中的callbacks参数见 docs/en/mkdocs.yml。二、一个带回调的应用发票场景让我们通过一个具体例子理解这一切。假设你开发了一个允许创建发票invoice的应用发票包含id、title可选、customer和total字段你 API 的用户一位外部开发者会通过 POST 请求在你的 API 中创建一张发票随后你的 API我们设想它会把发票发送给外部开发者的某个客户收款把通知发回给 API 用户外部开发者——这通过你的 API向该外部开发者提供的某个外部 API发送 POST 请求完成这就是回调callback。可以看到回调本质上就是一次普通的 HTTP 请求。真正的回调实现可能只有一两行代码例如callback_url https://example.com/api/v1/invoices/events/ httpx.post(callback_url, json{description: Invoice paid, paid: True})在实际实现回调时你可以使用 HTTPX 或 Requests 等 HTTP 客户端库。但回调中最重要的部分是确保你的 API 用户外部开发者能按照你的 API将在回调请求体中发送的数据正确地实现外部 API。因此接下来的重点不是实现回调本身而是编写文档代码说明这个外部 API应当如何设计才能接收来自你的 API的回调。这份文档会出现在你 API 的/docsSwagger UI 中让外部开发者知道如何构建外部 API。三、普通的 FastAPI 应用加入回调之前先看看添加回调之前这个 API 应用长什么样。它包含一个路径操作接收Invoice请求体以及一个承载回调 URL 的查询参数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_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}完整可运行代码见 docs_src/openapi_callbacks/tutorial001_py310.py。几个值得注意的细节callback_url查询参数使用了 Pydantic 的HttpUrl类型源码中为HttpUrl | None None这意味着 FastAPI 会校验它必须是合法的 URL同时允许缺省。整段代码中唯一的新东西是路径操作装饰器上的callbacksinvoices_callback_router.routes参数。我们将在下文详解。四、编写回调文档代码这段文档代码不会在你的应用中真正执行——我们只需要它来文档化那个外部 API应当长什么样。既然你已经知道如何用 FastAPI 轻松生成 API 的自动文档那么就用同样的知识去文档化外部 API创建外部 API 应当实现的路径操作也就是你的 API 将会调用的那些操作。编写技巧在写回调文档代码时可以想象你就是那位外部开发者正在实现外部 API而非你的 API。暂时采用这个视角会更容易判断参数、请求体 Pydantic 模型、响应模型等应该放在哪里。4.1 创建回调APIRouter首先创建一个新的APIRouter它将容纳一个或多个回调from fastapi import APIRouter, FastAPI from pydantic import BaseModel, HttpUrl invoices_callback_router APIRouter()4.2 创建回调路径操作回调的路径操作用上面创建的同一个APIRouter来定义它看起来就像普通的 FastAPI路径操作它应当声明自己要接收的请求体例如body: InvoiceEvent它也可以声明要返回的响应例如response_modelInvoiceEventReceived。class InvoiceEvent(BaseModel): description: str paid: bool class InvoiceEventReceived(BaseModel): ok: bool invoices_callback_router.post( {$callback_url}/invoices/{$request.body.id}, response_modelInvoiceEventReceived ) def invoice_notification(body: InvoiceEvent): pass与普通路径操作相比它有2 个主要区别不需要任何真实代码你的应用永远不会调用这段代码它仅用于文档化外部 API所以函数体可以只有pass路径可以包含 OpenAPI 3 表达式Key Expression详见下文其中可以使用变量来引用发送到你的 API的原始请求中的参数和组成部分。4.3 回调路径表达式OpenAPI 3 Key Expression回调的路径可以包含一个 OpenAPI 3 表达式Key Expression用于引用发送到你的 API的原始请求中的组成部分。在本例中这个路径表达式是{$callback_url}/invoices/{$request.body.id}它包含两种变量{$callback_url}引用原始请求中的查询参数callback_url{$request.body.id}引用原始请求 JSON 请求体中的id字段。我们走一遍完整流程。假设外部开发者向你的 API发送请求到https://yourapi.com/invoices/?callback_urlhttps://www.external.org/events携带如下 JSON 请求体{ id: 2expen51ve, customer: Mr. Richie Rich, total: 9999 }那么你的 API会处理这张发票并在稍后某个时刻向callback_url即外部 API发送回调请求https://www.external.org/events/invoices/2expen51ve携带的 JSON 请求体类似{ description: Payment celebration, paid: true }并且期待那个外部 API返回如下 JSON 响应体{ ok: true }注意观察最终的回调 URL 同时包含了查询参数callback_url传入的地址https://www.external.org/events以及 JSON 请求体内的发票id2expen51ve——这就是 OpenAPI 3 Key Expression 的威力把原始请求的任意参数拼接到回调目标路径中。关于 OpenAPI 3 Key Expression 的完整规范可查阅 OpenAPI Specification 3.1.0 的 Key Expression 章节本文示例仅用到$callback_url查询参数与$request.body.id请求体字段两种最常见形式。4.4 添加回调 router至此回调路径操作已经就绪也就是外部开发者应当在外部 API中实现的那一个。接下来在你的 API的路径操作装饰器中使用callbacks参数传入该回调 router 的.routes属性app.post(/invoices/, callbacksinvoices_callback_router.routes) def create_invoice(invoice: Invoice, callback_url: HttpUrl | None None): ...注意传给callbacks的不是 router 本身invoices_callback_router而是它的.routes即invoices_callback_router.routes。FastAPI 会用这些路由生成回调的 OpenAPI 文档。4.5 查看文档现在启动你的应用访问 http://127.0.0.1:8000/docs本地运行时你会看到文档中为你的路径操作增加了一个Callbacks区块展示外部 API应当如何实现从截图中可以看到截图源文件位于 docs/en/docs/img/tutorial/openapi-callbacks/image01.png在POST /invoices/操作下出现了Callbacks标签页回调项名为invoice_notification其下有一个POST子操作URL 为模板形式的{$callback_url}/invoices/{$request.body.id}描述为Invoice Notification该回调子操作声明了必需的请求体application/json示例{description: string, paid: true}以及200 Successful Response响应界面上的Try it out按钮说明 Swagger UI 甚至允许直接试调这个回调契约。五、源码级原理callbacks参数是如何生效的5.1 路由层参数接收与挂载callbacks是APIRoute以及APIRouter的各路径操作装饰器的一个正式参数类型为list[BaseRoute] | None见 fastapi/routing.py。在创建路由对象时它被直接挂载到路由实例上route.callbacks callbacks对应 fastapi/routing.py。也就是说回调路由集合与普通路径操作一样被作为路由元数据保存不会注册到实际的路由表中因此回调代码绝不会被执行——这与文档代码的定位完全一致。值得注意的是callbacks参数在APIRouter上同样存在例如 fastapi/routing.py 附近的 router 级定义注释明确说明这些回调应应用于该 router 下所有路径操作并可覆盖到所有端点。此外include_router与 context 合并逻辑中也有回调的传递与合并处理见 fastapi/routing.py 与 fastapi/routing.py说明回调可以跨 router 继承。5.2 OpenAPI 生成层如何序列化为callbacks字段真正把回调写进 OpenAPI 文档的逻辑位于 fastapi/openapi/utils.py当某个路径操作存在route.callbacks时FastAPI 会遍历回调路由列表对每个APIRoute类型的回调递归调用get_openapi_path()将其完整转换为一个 OpenAPI 路径项含请求体、响应、operationId 等然后以callbacks[callback.name] {callback.path: cb_path}的形式组装进operation[callbacks]。这意味着回调的键名是回调路径操作的函数名如invoice_notification回调的路径模板如{$callback_url}/invoices/{$request.body.id}会原样保留在 OpenAPI 文档中由下游工具如 Swagger UI 或代码生成器按 Key Expression 语义解析回调的请求体与响应模型会被纳入components/schemas成为可复用的 Schema 定义。同时在收集模型定义阶段fastapi/openapi/utils.py 会把回调路由涉及的字段模型如InvoiceEvent、InvoiceEventReceived一并收集进 flat models确保这些 Pydantic 模型会出现在最终的components中供回调路径项引用。5.3 测试验证OpenAPI Schema 的完整形态仓库中的测试 tests/test_tutorial/test_openapi_callbacks/test_tutorial001.py 对示例应用做了三件事test_get向/invoices/POST 请求断言返回{msg: Invoice received}证明普通路径操作照常工作test_dummy_callback直接调用invoice_notification({})仅为覆盖文档函数它本不会被应用调用test_openapi_schema断言/openapi.json的完整快照。从快照中可以看到最终生成的 OpenAPI 结构节选{ paths: { /invoices/: { post: { callbacks: { invoice_notification: { {$callback_url}/invoices/{$request.body.id}: { post: { summary: Invoice Notification, operationId: invoice_notification__callback_url__invoices___request_body_id__post, requestBody: { required: true, content: { application/json: { schema: {$ref: #/components/schemas/InvoiceEvent} } } }, responses: { 200: { description: Successful Response, content: { application/json: { schema: {$ref: #/components/schemas/InvoiceEventReceived} } } } } } } } } } } } }同时components.schemas中出现了Invoice、InvoiceEvent、InvoiceEventReceived等模型与文档中 Pydantic 模型的声明一一对应。这份测试快照既是自动化测试也是理解回调最终如何在 OpenAPI 中呈现的最佳参考。六、总结与实践要点回调的本质一次由你的 API发起、指向外部 API的普通 HTTP 请求OpenAPI Callbacks 只负责文档化这个外部契约不负责执行。实现三步走① 创建独立的APIRouter② 在 router 上定义回调路径操作函数体可为pass路径使用 OpenAPI 3 Key Expression③ 在你的 API的路径操作装饰器中传入callbackscallback_router.routes。Key Expression{$callback_url}引用查询参数、{$request.body.id}引用请求体字段可组合出动态回调 URL如{$callback_url}/invoices/{$request.body.id}。文档效果回调会以 Callbacks 区块出现在/docs的 Swagger UI 中并且会完整序列化进/openapi.json的callbacks字段参见 fastapi/openapi/utils.py可供外部开发者或代码生成工具直接消费。适用前提callbacks参数在APIRoute、APIRouter及include_router中均可使用见 fastapi/routing.py 与 fastapi/routing.py本文示例基于 Python 3.10 语法str | None若使用更低版本请改用Optional[...]。通过这一机制你的 API 文档不再是单方面的接口清单而是能够完整表达双向交互契约——你调用我我回调你两边都在 OpenAPI 中说得清清楚楚。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考