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 主体，以及一个 callback_url 查询参数，其中包含回调的 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"}

🤓 其他版本和变体

Python 3.9+

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，其中将包含一个或多个回调。

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"}

🤓 其他版本和变体

Python 3.9+

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。

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"}

🤓 其他版本和变体

Python 3.9+

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（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"}

🤓 其他版本和变体

Python 3.9+

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= 的不是路由器本身（invoices_callback_router），而是 .routes 属性，就像 invoices_callback_router.routes 一样。

检查文档¶

现在您可以启动您的应用并前往 http://127.0.0.1:8000/docs。

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

OpenAPI 回调¶

具有回调的应用¶

普通的 FastAPI 应用¶

记录回调¶

编写回调文档代码¶

创建一个回调 APIRouter¶

创建回调 *路径操作*¶

回调路径表达式¶

添加回调路由器¶

检查文档¶

创建一个回调 `APIRouter`¶

创建回调路径操作¶