FastAPI 与 OpenAPI Webhooks如何把「你的 API 反向调用用户」的契约写进 OpenAPI 文档【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi当你的应用需要主动通知对方在发生某种事件时携带数据调用对方的应用/接收请求就需要向 API 的使用者清楚地传达这一约定。这与通常「用户向你的 API 发请求」的方向相反——是你的 API你的应用向对方系统对方的 API、应用发请求。这种机制通常被称为Webhook。本文基于 FastAPI 官方文档 OpenAPI 的 Webhook 说明结合仓库源码与测试用例完整讲解如何用 FastAPI OpenAPI 把 Webhook 契约文档化并深入到源码层看清webhooks属性背后的实现。读完本文你能掌握Webhook 的典型协作流程以及它与普通 path operation 的本质区别用app.webhooks定义 Webhook 的完整可运行示例含 Pydantic 请求体模型生成的 OpenAPI JSON 中webhooks节点的具体结构以及它在源码中的生成路径为 Webhook 附加安全要求如 Bearer Token的写法。Webhook 的典型流程通常的步骤是定义事件数据首先在你自己的代码中定义要发送的消息也就是请求的正文body定义发送时机同时以某种方式定义应用发送这些请求事件的时机用户登记回调地址使用者定义应用应当把请求发送到哪个URL例如在某个 Web 仪表盘上登记。至于 Webhook URL 的登记方式、实际发送请求的代码等这些逻辑全部由你自行决定——用你自己的代码随意实现即可。FastAPI 本身不提供「发送 Webhook」的运行时机制它提供的核心价值在于把这套契约文档化让接收方知道你的应用会发什么、什么时候发、发到哪类地址。用 FastAPI 与 OpenAPI 文档化 Webhook使用FastAPI与 OpenAPI你可以定义Webhook 的名称应用可以发送的 HTTP操作例如POST、PUT等应用将发送的请求正文body。这样使用者要实现一个接收你的Webhook请求的API就大大方便了。在某些情况下使用者甚至可以基于这份 OpenAPI 文档自动生成他们 API 侧的代码。备注Webhook 需要 OpenAPI 3.1.0 及以上版本FastAPI 从0.99.0起支持。当前仓库源码中 OpenAPI 输出默认即为3.1.0见 openapi/utils.py 中 get_openapi 的默认参数。带 Webhook 的应用创建FastAPI应用时会有一个webhooks属性。你可以在上面像定义path operation一样定义webhook例如app.webhooks.post()。仓库中的官方示例 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 well 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]要点拆解class Subscription(BaseModel)用 Pydantic 模型定义 Webhook 请求的 body 结构——username字符串、monthly_fee浮点数、start_date日期时间。OpenAPI 会据此生成完整的 schemaapp.webhooks.post(new-subscription)声明一个 Webhook其 HTTP 操作为POST标识符为new-subscription端点函数的 docstring 会被原样写入 OpenAPI 的description字段是向使用者说明「什么时候会收到这个请求」的最佳位置。运行应用后启动即可uvicorn main:app --reload把上面代码保存为main.py然后访问http://127.0.0.1:8000/docs查看文档。源码佐证app.webhooks就是一个 APIRouter备注app.webhooks对象实际上只是一个普通的APIRouter与你在多文件组织应用时使用的 router 是同一类型。这一点可以直接在源码中确认。applications.py 中FastAPI.__init__把webhooks参数或其默认值赋给了self.webhooksself.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()即如果你在FastAPI(webhooksapi_webhooks)构造时传入了一个APIRouter就使用它否则自动创建一个新的APIRouter()。正因如此app.webhooks.post()、app.webhooks.get()等装饰器与普通APIRouter的方法完全一致——你可以在独立的模块文件里先定义webhooks APIRouter()批量声明全部事件后再传入FastAPI(webhookswebhooks)实现与多文件组织应用相同的模块化拆分。构造参数webhooks的官方文档注释也说明了它与callbacks的关系「This is similar tocallbacksbut it doesnt depend on specificpath operations.」见 applications.py。注意Webhook 名是「标识符」不是 URL 路径Webhook 并没有声明路径不像/items/那样。这里传入的字符串是 webhook 的标识符事件名。例如app.webhooks.post(new-subscription)中webhook 的名字就是new-subscription。之所以这样设计是因为使用者实际希望接收 Webhook 请求的URL 路径需要通过另外的方式例如 Web 仪表盘自行定义——每个使用者注册的回调地址都可能不同但「事件名 → 请求结构」的契约是固定的。检查文档/docs 中的 Webhook 区块启动应用并访问 http://127.0.0.1:8000/docs可以看到文档中除了常规的path operations之外还会显示webhooks区块如上图。上图中new-subscription作为独立条目列出展开后可以查看它的POST操作、请求体 schema 与描述说明。与此同时/openapi.json会生成如下结构以下节选自 tests/test_tutorial/test_openapi_webhooks/test_tutorial001.py 中对/openapi.json响应的快照断言{ openapi: 3.1.0, info: {title: FastAPI, version: 0.1.0}, paths: { /users/: { get: { summary: Read Users, operationId: read_users_users__get, responses: { 200: { description: Successful Response, content: {application/json: {schema: {}}} } } } } }, webhooks: { new-subscription: { post: { summary: New Subscription, description: When a new user subscribes to your service well send you a POST request with this\ndata to the URL that you register for the event new-subscription in the dashboard., operationId: new_subscriptionnew_subscription_post, requestBody: { content: { application/json: { schema: {$ref: #/components/schemas/Subscription} } }, required: true }, responses: { 200: { description: Successful Response, content: {application/json: {schema: {}}} }, 422: { description: Validation Error, content: { application/json: { schema: {$ref: #/components/schemas/HTTPValidationError} } } } } } } }, components: { schemas: { Subscription: { properties: { username: {type: string, title: Username}, monthly_fee: {type: number, title: Monthly Fee}, start_date: { type: string, format: date-time, title: Start Date } }, type: object, required: [username, monthly_fee, start_date], title: Subscription } } } }从这份快照可以看出几个关键事实Webhook 以事件名为键挂在顶层webhooks节点下new-subscription与常规路径所在的paths节点平级——这正是「标识符而非路径」在 OpenAPI 层面的体现Subscription模型被收集进components.schemaswebhook 的requestBody通过$ref引用它端点函数的 docstring 完整保留在description中包含换行。源码级实现Webhook 路由如何进入 OpenAPI从源码结构看Webhook 文档化的完整调用链如下收集路由FastAPI实例的openapi属性在生成或重新生成schema 时把常规路由和 Webhook 路由分别传给get_openapi()——routesself.routes与webhooksself.webhooks.routes见 applications.py。注意app.webhooks本身并不会挂载到 ASGI 应用的路由表上它只被用于 OpenAPI 生成所以 Webhook 端点函数不会被 HTTP 请求触发合并字段收集openapi/utils.py 中get_openapi()把routes与webhooks的所有字段合在一起做模型收集get_fields_from_routes(list(routes) list(webhooks or []))因此 Webhook 用到的 Pydantic 模型会正常进入components.schemas与常规路径共享同一套 schema单独生成 webhooks 节点常规路由的 path 进入pathsWebhook 路由则在独立的循环中处理按api_webhook.path_format即事件名聚合最终只有当webhook_paths非空时才写入output[webhooks]见 openapi/utils.pyfor 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) ... ... output[paths] paths if webhook_paths: output[webhooks] webhook_paths可以看到 Webhook 与常规路由复用同一套get_openapi_path()逻辑——因此 summary、operationId、requestBody、responses 等 OpenAPI 细节与普通 path operation 的生成规则完全一致 4.数据模型承载输出结构由 OpenAPI 模型校验webhooks是合法的顶层字段见 openapi/models.pywebhooks: dict[str, PathItem | Reference] | None None这也解释了前文快照中webhooks为何是「事件名 → PathItem」的字典结构。进阶为 Webhook 附加安全要求Webhook 端点声明虽然不会被真实触发但它完整支持 FastAPI 的参数与安全体系可以把「接收方必须如何鉴权」也写进契约。tests/test_webhooks_security.py 展示了给 Webhook 附加HTTPBearer安全要求的写法bearer_scheme HTTPBearer() app.webhooks.post(new-subscription) def new_subscription( body: Subscription, token: Annotated[str, Security(bearer_scheme)] ): When a new user subscribes to your service well send you a POST request with this data to the URL that you register for the event new-subscription in the dashboard. 对应的 OpenAPI 输出会在该 Webhook 操作节点上生成security: [{HTTPBearer: []}]并在components.securitySchemes中登记{HTTPBearer: {type: http, scheme: bearer}}见 tests/test_webhooks_security.py 的快照断言。使用者据此就知道接收这个 Webhook 时需要在请求中携带 Bearer Token。小结FastAPI 通过app.webhooks本质是一个APIRouter见 applications.py让你以声明 path operation 的方式声明 Webhook事件名作为标识符、HTTP 方法作为操作、Pydantic 模型作为请求体契约生成 OpenAPI 时Webhook 路由与常规路由共享字段收集与路径生成逻辑最终挂在顶层webhooks节点下openapi/utils.py、openapi/models.py并在 Swagger UI 的/docs中直观展示Webhook 端点不参与实际路由分发其价值在于文档与契约——配合Security等能力还可以把回调鉴权要求一并交给使用者该能力要求 OpenAPI 3.1.0 与 FastAPI 0.99.0当前仓库的默认输出即为 3.1.0。可继续在仓库中深入的文件官方示例 docs_src/openapi_webhooks/tutorial001_py310.py、快照测试 tests/test_tutorial/test_openapi_webhooks/test_tutorial001.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），仅供参考