OpenAPI 回调

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

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

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

带有回调的应用

让我们通过一个例子来看看所有这些。

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

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

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

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

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

正常的 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"}

Tip

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 的 Swagger UI 中显示在 /docs，并让外部开发者知道如何构建*外部 API*。

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

Tip

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

当你自己实现回调时，你可以使用类似 HTTPX 或 Requests 的工具。

编写回调文档代码

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

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

因此，我们将使用相同的知识来记录*外部 API* 应该是什么样子... 通过创建*外部 API* 应该实现的路径操作（你的 API 将调用的那些）。

Tip

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

暂时采用这种观点（作为*外部开发者*）可以帮助你感觉更明显地知道在哪里放置参数、主体的 Pydantic 模型、响应的 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"}
  

  与普通的*路径操作*相比，主要有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
}

Tip

注意回调URL如何包含作为查询参数接收的URL callback_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"}

Tip

注意，你不是将路由器本身（invoices_callback_router）传递给 callback=，而是传递属性 .routes，如 invoices_callback_router.routes。

查看文档

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

你将看到你的文档中包含一个“Callbacks”部分，用于你的*路径操作*，展示了*外部API*应该是什么样子：