三周精通FastAPI：36 OpenAPI 回调

官方文档：

OpenAPI 回调¶

您可以创建触发外部 API 请求的路径操作 API，这个外部 API 可以是别人创建的，也可以是由您自己创建的。

API 应用调用外部 API 时的流程叫做回调。因为外部开发者编写的软件发送请求至您的 API，然后您的 API 要进行回调，并把请求发送至外部 API。

此时，我们需要存档外部 API 的信息，比如应该有哪些路径操作，返回什么样的请求体，应该返回哪种响应等。

使用回调的应用¶

示例如下。

假设要开发一个创建发票的应用。

发票包括 id、title（可选）、customer、total 等属性。

API 的用户（外部开发者）要在您的 API 内使用 POST 请求创建一条发票记录。

（假设）您的 API 将：

把发票发送至外部开发者的消费者
归集现金
把通知发送至 API 的用户（外部开发者）
- 通过（从您的 API）发送 POST 请求至外部 API （即回调）来完成

常规 FastAPI 应用¶

添加回调前，首先看下常规 API 应用是什么样子。

常规 API 应用包含接收 Invoice 请求体的路径操作，还有包含回调 URL 的查询参数 callback_url。这部分代码很常规，您对绝大多数代码应该都比较熟悉了：

from typing import Unionfrom fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrlapp = FastAPI()class Invoice(BaseModel):id: strtitle: Union[str, None] = Nonecustomer: strtotal: floatclass InvoiceEvent(BaseModel):description: strpaid: boolclass InvoiceEventReceived(BaseModel):ok: boolinvoices_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: Union[HttpUrl, None] = None):"""Create an invoice.This will (let's imagine) let the API user (some external developer) create aninvoice.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 theexternal 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"}

"提示"

callback_url 查询参数使用 Pydantic 的 URL 类型。

此处唯一比较新的内容是路径操作装饰器中的 callbacks=invoices_callback_router.routes参数，下文介绍。

存档回调¶

实际的回调代码高度依赖于您自己的 API 应用。

并且可能每个应用都各不相同。回调代码可能只有一两行，比如：

callback_url = "https://example.com/api/v1/invoices/events/"
requests.post(callback_url, json={"description": "Invoice paid", "paid": True})

但回调最重要的部分可能是，根据 API 要发送给回调请求体的数据等内容，确保您的 API 用户（外部开发者）正确地实现外部 API。

因此，我们下一步要做的就是添加代码，为从 API 接收回调的外部 API存档。

这部分文档在 /docs 下的 Swagger API 文档中显示，并且会告诉外部开发者如何构建外部 API。

本例没有实现回调本身（只是一行代码），只有文档部分。

"提示"

实际的回调只是 HTTP 请求。

实现回调时，要使用 HTTPX 或 Requests。

编写回调文档代码¶

应用不执行这部分代码，只是用它来记录外部 API 。

但，您已经知道用 FastAPI 创建自动 API 文档有多简单了。

我们要使用与存档外部 API 相同的知识……通过创建外部 API 要实现的路径操作（您的 API 要调用的）。

"提示"

编写存档回调的代码时，假设您是外部开发者可能会用的上。并且您当前正在实现的是外部 API，不是您自己的 API。

临时改变（为外部开发者的）视角能让您更清楚该如何放置外部 API 响应和请求体的参数与 Pydantic 模型等。

创建回调的 `APIRouter`¶

首先，新建包含一些用于回调的 APIRouter。

from fastapi import APIRouter, FastAPIinvoices_callback_router = APIRouter()

创建回调路径操作¶

创建回调路径操作也使用之前创建的 APIRouter。

它看起来和常规 FastAPI 路径操作差不多：

声明要接收的请求体，例如，body: InvoiceEvent
还要声明要返回的响应，例如，response_model=InvoiceEventReceived

class InvoiceEvent(BaseModel):description: strpaid: boolclass InvoiceEventReceived(BaseModel):ok: boolinvoices_callback_router = APIRouter()@invoices_callback_router.post("{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
)
def invoice_notification(body: InvoiceEvent):pass

回调路径操作与常规路径操作有两点主要区别：

它不需要任何实际的代码，因为应用不会调用这段代码。它只是用于存档外部 API。因此，函数的内容只需要 pass 就可以了
路径可以包含 OpenAPI 3 表达式（详见下文），可以使用带参数的变量，以及发送至您的 API 的原始请求的部分

回调路径表达式¶

回调路径支持包含发送给您的 API 的原始请求的部分的 OpenAPI 3 表达式。

本例中是字符串：

"{$callback_url}/invoices/{$request.body.id}"

因此，如果您的 API 用户（外部开发者）发送请求到您的 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）。

添加回调路由¶

至此，在上文创建的回调路由里就包含了回调路径操作（外部开发者要在外部 API 中实现）。现在使用 API 路径操作装饰器的参数 callbacks，从回调路由传递属性 .routes（实际上只是路由/路径操作的列表）：

@app.post("/invoices/", callbacks=invoices_callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: Union[HttpUrl, None] = None):

"提示"

注意，不能把路由本身（invoices_callback_router）传递给 callback=，要传递 invoices_callback_router.routes 中的 .routes 属性。

查看文档¶

现在，使用 Uvicorn 启动应用，打开 http://127.0.0.1:8000/docs。

就能看到文档的路径操作已经包含了回调的内容以及外部 API：

实践

源代码

将代码写入callback.py文件

from typing import Unionfrom fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrlapp = FastAPI()class Invoice(BaseModel):id: strtitle: Union[str, None] = Nonecustomer: strtotal: floatclass InvoiceEvent(BaseModel):description: strpaid: boolclass InvoiceEventReceived(BaseModel):ok: boolinvoices_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: Union[HttpUrl, None] = None):"""Create an invoice.This will (let's imagine) let the API user (some external developer) create aninvoice.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 theexternal 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"}

启动服务

uvicorn callback:app --reload

测试

curl命令：

curl -X 'POST' \'http://127.0.0.1:8000/invoices/?callback_url=http%3A%2F%2Fwww.quye.com' \-H 'accept: application/json' \-H 'Content-Type: application/json' \-d '{"id": "string","title": "string","customer": "string","total": 0
}'

返回信息：

{"msg":"Invoice received"}

证明回调成功。若回调失败，会给出更详细的信息：

{"detail":[{"type":"url_parsing","loc":["query","callback_url"],"msg":"Input should be a valid URL, relative URL without a base","input":"quye.com","ctx":{"error":"relative URL without a base"}}]}