OpenAPI 回调¶

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

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

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

带回调的应用¶

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

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

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

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

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

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

普通的 FastAPI 应用¶

让我们先看看在添加回调之前，普通的 API 应用是什么样子的。

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

这部分非常普通，大部分代码你可能已经很熟悉了。

Python 3.10+

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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，其中将包含一个或多个回调。

Python 3.10+

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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。

Python 3.10+

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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 的响应体类似于

{
    "ok": true
}

提示

注意所使用的回调 URL 如何包含作为查询参数在 callback_url 中接收到的 URL（https://www.external.org/events），以及 JSON 请求体内部的发票 id（2expen51ve）。

添加回调路由¶

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

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

Python 3.10+

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

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: 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: 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。

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

OpenAPI 回调¶

带回调的应用¶

普通的 FastAPI 应用¶

记录回调文档¶

编写回调文档代码¶

创建一个回调 APIRouter¶

创建回调 路径操作¶

回调路径表达式¶

添加回调路由¶

检查文档¶

创建一个回调 `APIRouter`¶

创建回调路径操作¶