跳到内容

OpenAPI 回调

你可以创建一个带 *路径操作* 的 API，它可能会触发对由其他人（很可能是将 *使用* 你的 API 的同一开发者）创建的 *外部 API* 的请求。

当你的 API 应用程序调用 *外部 API* 时发生的过程称为“回调”。因为外部开发者编写的软件向你的 API 发送请求，然后你的 API *回调*，向一个 *外部 API*（很可能是由同一开发者创建的）发送请求。

在这种情况下，你可能希望记录该外部 API *应该* 是什么样子。它应该有什么 *路径操作*，它应该期望什么请求体，它应该返回什么响应等等。

带回调的应用程序

让我们通过一个例子来看这一切。

想象你开发一个允许创建发票的应用程序。

这些发票将包含 `id`、`title`（可选）、`customer` 和 `total`。

你的 API 用户（外部开发者）将通过 POST 请求在你的 API 中创建发票。

然后你的 API 将（让我们想象）

  • 将发票发送给外部开发者的某些客户。
  • 收款。
  • 向 API 用户（外部开发者）发送通知。
    • 这将通过向该外部开发者提供的某个 *外部 API* 发送 POST 请求（从 *你的 API*）来完成（这就是“回调”）。

正常的 FastAPI 应用程序

让我们先看看在添加回调之前，正常的 API 应用程序会是什么样子。

它将有一个 *路径操作*，接收一个 `Invoice` 请求体，以及一个查询参数 `callback_url`，其中包含回调的 URL。

这部分很正常，大部分代码你可能已经很熟悉了

from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[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: Union[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"}

提示

callback_url 查询参数使用了 Pydantic Url 类型。

唯一的新东西是 `callbacks=invoices_callback_router.routes` 作为 *路径操作装饰器* 的参数。我们接下来会看到那是什么。

文档化回调

实际的回调代码将很大程度上取决于你自己的 API 应用程序。

它也可能在不同应用程序之间差异很大。

它可能只有一两行代码，比如

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

但回调中最重要的部分可能是确保你的 API 用户（外部开发者）根据 *你的 API* 将在回调请求体中发送的数据等正确实现 *外部 API*。

所以，我们接下来要做的是添加代码来记录该 *外部 API* 应该是什么样子，以便接收来自 *你的 API* 的回调。

该文档将显示在你的 API 的 `/docs` 处的 Swagger UI 中，它将让外部开发者知道如何构建 *外部 API*。

这个例子没有实现回调本身（那可能只是一行代码），只实现了文档部分。

提示

实际的回调只是一个 HTTP 请求。

当你自己实现回调时，可以使用像 HTTPX 或 Requests 这样的库。

编写回调文档代码

这段代码不会在你的应用程序中执行，我们只需要它来 *文档化* 该 *外部 API* 应该是什么样子。

但是，你已经知道如何使用 FastAPI 轻松为 API 创建自动文档。

所以我们将使用相同的知识来文档化 *外部 API* 应该是什么样子……通过创建外部 API 应该实现的 *路径操作*（也就是你的 API 将调用的那些）。

提示

在编写文档化回调的代码时，想象你是那个 *外部开发者* 可能会很有用。并且你当前正在实现 *外部 API*，而不是 *你的 API*。

暂时采纳这种观点（*外部开发者* 的观点）可以帮助你更容易地了解该 *外部 API* 的参数、请求体 Pydantic 模型、响应等应该放在哪里。

创建回调 APIRouter

首先创建一个新的 APIRouter，它将包含一个或多个回调。

from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[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: Union[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"}

创建回调 *路径操作*

要创建回调 *路径操作*，使用你上面创建的同一个 APIRouter。

它看起来应该就像一个正常的 FastAPI *路径操作*

  • 它可能应该有一个它将接收的请求体的声明，例如 `body: InvoiceEvent`。
  • 它还可以有一个它应该返回的响应的声明，例如 `response_model=InvoiceEventReceived`。
from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[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: Union[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"}

与正常的 *路径操作* 有 2 个主要区别

  • 它不需要有任何实际代码，因为你的应用程序永远不会调用这段代码。它仅用于文档化 *外部 API*。因此，该函数可以只包含 `pass`。
  • *路径* 可以包含一个 OpenAPI 3 表达式（详见下文），其中可以使用包含参数的变量以及发送到 *你的 API* 的原始请求的部分内容。

回调路径表达式

回调 *路径* 可以包含一个 OpenAPI 3 表达式，其中可以包含发送到 *你的 API* 的原始请求的部分内容。

在这种情况下，它是 `str`

"{$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` 中作为查询参数接收的 URL (`https://www.external.org/events`)，以及来自 JSON 请求体内部的发票 `id` (`2expen51ve`)。

添加回调路由器

此时，你已经在上面创建的回调路由器中拥有了所需的回调 *路径操作*（即 *外部开发者* 应该在 *外部 API* 中实现的那些）。

现在，使用 *你的 API 的路径操作装饰器* 中的 `callbacks` 参数，传递该回调路由器的 `.routes` 属性（这实际上只是一个路由/*路径操作* 的 `list`）

from typing import Union

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: Union[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: Union[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"}

提示

请注意，你不是将路由器本身 (`invoices_callback_router`) 传递给 `callback=`，而是传递 `.routes` 属性，例如 `invoices_callback_router.routes`。

检查文档

现在你可以启动你的应用程序并访问 http://127.0.0.1:8000/docs。

你将看到你的文档，其中包括你的 *路径操作* 的“回调”部分，该部分显示了 *外部 API* 应该是什么样子