FastAPI OpenAPI Webhooks 文档化指南:用 app.webhooks 声明式描述你的应用将主动推送的事件
【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi
本篇指南聚焦 FastAPI 的OpenAPI Webhooks特性:在 API 的用户需要反向接收你的应用推送通知的场景下,如何用app.webhooks在 OpenAPI Schema 与自动生成的文档界面中声明这些事件(事件名、HTTP 方法、请求体)。读完后你可以完整掌握 Webhooks 的工作流程、可复制的示例代码,以及 Webhooks 在 OpenAPI 输出与文档界面中的呈现方式,并能深入理解其源码级实现原理。
Webhooks 是什么:请求方向的"反转"
在常规的 API 交互中,是你的用户(客户端)向你的 API发送请求。但在某些场景下,流程恰好相反:你的应用(或你的 API)需要向用户的系统(用户的 API、用户的应用)发送请求,通常是为了通知某个特定**事件(Event)**的发生。
这种模式通常被称为Webhook(网络钩子)。典型例子:
- 支付平台在订单支付完成后,向商家回调 URL 推送"支付成功"通知;
- SaaS 服务在用户新订阅、退订、升级套餐时,向客户注册的 URL 推送事件数据。
核心特征是:目标 URL 不是你的 API 路由,而是你的用户在别处(例如他们自己的 Dashboard)配置并登记的地址,由你的代码在适当时机主动向该地址发起请求。
Webhooks 的工作流程:三方各自负责什么
一个完整的 Webhook 机制通常由三方协作完成,理解各自职责有助于正确设计 API:
- 你在代码中定义消息体(Request body):明确你希望发送的消息结构,即请求体长什么样。这是你 API 契约的一部分,也是需要在 OpenAPI 中文档化的核心内容。
- 你定义发送时机(触发事件):在你的应用中以某种方式定义"在哪些时刻"这些请求/事件会被发出,例如"新用户完成订阅后"。
- 你的用户定义接收 URL:你的用户通过某种方式(通常是在一个 Web Dashboard 中)登记他们的接收端点 URL,你的应用将把请求发送到该 URL。
需要注意的一点是:Webhook 的 URL 注册逻辑与实际发送请求的代码,全部由你自己实现。FastAPI(以及 OpenAPI 规范)提供的是"文档与契约描述"能力,即告诉你的用户"我会发送什么、什么时候发送",而具体如何存储用户注册的 URL、何时触发 HTTP 调用,是你在自己业务代码中自由编写的。
用 FastAPI 与 OpenAPI 文档化 Webhooks
借助 FastAPI 对 OpenAPI 的支持,你可以在 API 文档中声明:
- 这些 Webhooks 的名称(事件标识,如
new-subscription); - 你的应用可能发送的HTTP 操作类型(如
POST、PUT等); - 你的应用将会发送的Request body结构(完整的 JSON Schema)。
这样做能让你的用户更简单地实现他们那侧的接收端点——他们可以直接阅读你的 OpenAPI 文档甚至自动生成的接口契约,在自己的系统中生成接收代码,而不需要靠口头约定或截图。
版本要求:Webhooks 是OpenAPI 3.1.0 及以上规范中的特性,从FastAPI
0.99.0开始支持。由于当前仓库中 get_openapi 的默认openapi_version参数就是"3.1.0",因此默认生成的 Schema 版本即可承载 Webhooks 字段。
完整示例:一个带 Webhooks 的应用
下面是一个完整可运行的示例,源码位于 docs_src/openapi_webhooks/tutorial001_py310.py:
from datetime import datetime from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Subscription(BaseModel): username: str monthly_fee: float start_date: datetime @app.webhooks.post("new-subscription") def new_subscription(body: Subscription): """ When a new user subscribes to your service we'll send you a POST request with this data to the URL that you register for the event `new-subscription` in the dashboard. """ @app.get("/users/") def read_users(): return ["Rick", "Morty"]示例要点逐一拆解:
Subscription模型:定义了 Webhook 请求体的结构(username、monthly_fee、start_date三个必填字段)。这个 Pydantic 模型会被 FastAPI 转成 OpenAPI 的组件 Schema(#/components/schemas/Subscription),你的用户据此就能知道每个字段的数据类型与是否必填。@app.webhooks.post("new-subscription"):这就是定义 Webhook 的装饰器,用法与你写普通路径操作(@app.post()等)几乎一致,只是挂在app.webhooks这个特殊属性上。- 文档字符串:函数的 docstring 会作为 Webhook 的
description出现在 OpenAPI Schema 中,向用户解释"触发时机 + 发送内容 + URL 在哪登记",这是面向用户的关键说明。 /users/常规路径操作:与 Webhook 定义共存,用于演示文档中两类条目的并列呈现。
app.webhooks就是一个APIRouter
app.webhooks对象实际上就是一个APIRouter——与你把大型应用拆分成多文件、多路由模块时使用的类型完全相同。这一点可以从 FastAPI 应用类源码 得到印证:
self.webhooks: Annotated[ routing.APIRouter, Doc( """ The `app.webhooks` attribute is an `APIRouter` with the *path operations* that will be used just for documentation of webhooks. ... """ ), ] = webhooks or routing.APIRouter()这意味着你可以在app.webhooks上使用APIRouter的全部能力:同样的装饰器风格、同样的 Pydantic 请求体解析、同样的依赖注入(dependencies)与安全方案声明,只是这些路由仅用于文档,不会被注册到实际应用的路由树中。
Webhook 的"路径"其实只是一个标签
定义 Webhook 时注意:你并没有声明一个真实的路径(如/items/)。装饰器中传入的字符串(如"new-subscription")只是这个 Webhook 的标识(事件名),在@app.webhooks.post("new-subscription")中,new-subscription就是 Webhook 名称。
之所以这样设计,是因为真正的 URL 路径预期由你的用户以其他方式定义(例如在他们的 Dashboard 中为每个事件登记接收地址)。你的应用只需承诺"当new-subscription事件发生时,我会 POST 这样一个请求体",而"POST 到哪个 URL"由用户侧数据驱动。
测试文档效果
在示例目录中启动应用并访问http://127.0.0.1:8000/docs:
uvicorn tutorial001_py310:app --reload打开文档界面后,你会看到与普通路径操作并列的Webhooks分组(如上文配图所示):
- 常规部分显示
GET /users/路径操作; - Webhooks 部分显示
POST new-subscription,展开后包含描述文本、无参数的说明、Request body(required)的示例值与 Schema 标签页,以及 Responses(200、422)信息。
你的用户可以直接在这个界面中查看事件契约,甚至基于/openapi.json做自动化处理。
源码级实现:Webhooks 如何进入 OpenAPI Schema
1. 路由上下文统一处理
在 get_openapi 函数中,普通路由与 Webhooks 被一起纳入字段收集与模型定义生成:
webhook_paths: dict[str, dict[str, Any]] = {} ... all_fields = get_fields_from_routes(list(routes) + list(webhooks or [])) ... for webhook_context in routing.iter_route_contexts(webhooks or []): api_webhook = _get_api_route_for_openapi(webhook_context) if api_webhook is not None: result = get_openapi_path(...) if result: path, security_schemes, path_definitions = result if path: webhook_paths.setdefault(api_webhook.path_format, {}).update(path) ... if webhook_paths: output["webhooks"] = webhook_paths可以看到:Webhook 路由与常规路由走同一套get_openapi_path处理逻辑(参数解析、请求体 Schema、响应声明),只是最终结果被写入独立的webhook_paths字典,并作为顶层webhooks字段输出——这正是 OpenAPI 3.1.0 对 Webhooks 的规定位置。只有当至少定义了一个 Webhook 时,webhooks键才会出现在输出中。
2. Schema 模型中的webhooks字段
在 OpenAPI 文档模型 fastapi/openapi/models.py 中,webhooks被声明为:
webhooks: dict[str, PathItem | Reference] | None = None即:一个从**事件名到PathItem(或引用)**的字典。这与 OpenAPI 规范中webhooks对象的结构一致:键是 Webhook 名称,值描述该事件下各 HTTP 方法的操作定义。
3. 从测试快照看真实输出结构
测试 tests/test_webhooks_security.py 断言了完整的/openapi.json输出,可以据此确认 Webhooks 在 Schema 中的真实形态:
{ "openapi": "3.1.0", "info": {"title": "FastAPI", "version": "0.1.0"}, "paths": {}, "webhooks": { "new-subscription": { "post": { "summary": "New Subscription", "description": "When a new user subscribes to your service ...", "operationId": "new_subscriptionnew_subscription_post", "requestBody": { "content": { "application/json": { "schema": {"$ref": "#/components/schemas/Subscription"} } }, "required": true }, "responses": { "200": {"description": "Successful Response", "...": "..."}, "422": {"description": "Validation Error", "...": "..."} }, "security": [{"HTTPBearer": []}] } } }, "components": { "schemas": {"Subscription": {"...": "..."}}, "securitySchemes": {"HTTPBearer": {"type": "http", "scheme": "bearer"}} } }这份快照还揭示了一个实用细节:Webhook 操作可以声明安全方案。测试中的应用示例为 Webhook 加上了HTTPBearer安全依赖:
bearer_scheme = HTTPBearer() @app.webhooks.post("new-subscription") def new_subscription( body: Subscription, token: Annotated[str, Security(bearer_scheme)] ): """..."""由于 Webhook 请求体中包含敏感数据(用户名、费用等),你完全可以在文档中向用户声明"发送该请求时会携带 Bearer Token",并让components.securitySchemes一并输出对应的方案定义(如{"HTTPBearer": {"type": "http", "scheme": "bearer"}}),让你的用户知道需要校验什么凭证。这在支付、订阅等涉及敏感数据的回调场景中尤其有价值。
小结:文档是 Webhook 契约的一半
回顾本文的核心要点:
- Webhooks 是请求方向的反转:你的应用向用户的系统发送请求以通知事件,接收 URL 由用户侧登记,发送逻辑由你的业务代码自行实现。
- FastAPI 提供契约文档能力:通过
@app.webhooks.post("事件名")这类装饰器,把事件名、HTTP 方法、请求体 Schema、描述甚至安全方案声明进 OpenAPI Schema 与文档界面,让用户可以据此实现(甚至自动生成)接收端点。 - 适用前提:需要 OpenAPI 3.1.0 及以上规范、FastAPI
0.99.0及以上版本;app.webhooks本质是APIRouter,其路由仅用于文档,不进入应用真实路由树;Webhook 的"路径"参数只是事件标识,而非真实 URL 路径。
相关仓库路径供深入阅读:示例源码 docs_src/openapi_webhooks/tutorial001_py310.py、Webhook 属性定义 fastapi/applications.py、OpenAPI 生成逻辑 fastapi/openapi/utils.py、Schema 模型 fastapi/openapi/models.py、含安全方案的测试 tests/test_webhooks_security.py。
【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考