跳到内容

OpenAPI 回调

您可以创建一个带有路径操作的 API，该路径操作可以触发对外部 API的请求，该外部 API是由其他人创建的（可能是与使用您的 API 的同一个开发人员）。

当您的 API 应用程序调用外部 API时发生的进程称为“回调”。因为外部开发人员编写的软件向您的 API 发送请求，然后您的 API回拨，向外部 API（可能由同一个开发人员创建）发送请求。

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

带有回调的应用程序

让我们用一个例子来了解所有这些。

假设您开发了一个允许创建发票的应用程序。

这些发票将有一个 id、title（可选）、customer 和 total。

您的 API 的用户（外部开发人员）将使用 POST 请求在您的 API 中创建一个发票。

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

  • 将发票发送给外部开发人员的某个客户。
  • 收取款项。
  • 将通知发送回 API 用户（外部开发人员）。
    • 这将通过从您的 API发送 POST 请求（到外部开发人员提供的外部 API，这就是“回调”）来完成。

普通的 FastAPI 应用程序

让我们首先看看在添加回调之前普通 API 应用程序的样子。

它将有一个路径操作，它将接收一个 Invoice 主体，以及一个包含回调 URL 的查询参数 callback_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*。

暂时采用这种观点（*外部开发人员*）可以帮助您感觉到将参数、Pydantic 模型放在哪里对于 *外部 API* 的主体、响应等更加明显。

创建一个回调 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"}

与正常的 *路径操作* 相比，有两个主要区别。

  • 它不需要有任何实际的代码，因为您的应用程序永远不会调用此代码。它仅用于记录 *外部 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* 应该是什么样子。