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_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"}完整可运行代码见 docs_src/openapi_callbacks/tutorial001_py310.py。
几个值得注意的细节:
callback_url查询参数使用了 Pydantic 的HttpUrl类型(源码中为HttpUrl | None = None),这意味着 FastAPI 会校验它必须是合法的 URL,同时允许缺省。- 整段代码中唯一的新东西,是路径操作装饰器上的
callbacks=invoices_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_model=InvoiceEventReceived。
class InvoiceEvent(BaseModel): description: str paid: bool class InvoiceEventReceived(BaseModel): ok: bool @invoices_callback_router.post( "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived ) 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_url=https://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 请求体内的发票id(2expen51ve)——这就是 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/", callbacks=invoices_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的路径操作装饰器中传入callbacks=callback_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),仅供参考