FastApi教程|OpenAPI回调

发表时间：2020-03-09

您可以使用 路径操作 创建一个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。

这部分很正常，您可能已经熟悉了大多数代码： 

from fastapi import APIRouter, FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: str = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter(default_response_class=JSONResponse)


@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):
    """
    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=messages_callback_router.routes 用作 路径操作装饰器 的参数 。 我们将看到接下来的内容。

记录回调

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

从一个应用程序到下一个应用程序，它可能会有很大的不同。

它可能只是一两行代码，例如： 

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

但是，回调中最重要的部分可能是确保您的API用户（外部开发人员） 根据 您的API 将在回调的请求正文中发送 的数据正确 实现 外部API 。

因此，我们接下来要做的是添加代码以记录该 外部API 看起来应 如何 从 您的API 接收回调 。

该文档将显示 /docs 在您API 的Swagger UI 中，并使外部开发人员知道如何构建 外部API 。

此示例未实现回调本身（可能只是一行代码），仅实现了文档部分。

小费

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

自己实现回调时，可以使用 HTTPX 或 Requests之 类的东西 。

编写回调文档代码

此代码将不会在你的应用程序来执行，我们只需要它 的文档 该如何 外部API 应该像。

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

因此，我们将使用相同的知识来记录 外部API的 外观……通过创建 外部API应该实现 的 路径操作（ 您的API将调用 的 路径操作） 来实现。

小费

在编写用于记录回调的代码时，想象您是该 外部开发人员 可能会很有用 。 并且您当前正在实现 外部API ，而不是 您的API 。

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

创建一个回调 APIRouter

首先创建一个 APIRouter 包含一个或多个回调的新代码。

该路由器永远不会添加到实际 FastAPI 应用中（即，永远不会传递到 app.include_router(...) ）。

因此，您需要声明 default_response_class ，并将其设置为 JSONResponse 。

技术细节

在 response_class 通常由设置 FastAPI 应用程序的调用过程 app.include_router(some_router) 。

但是由于我们从不打电话 app.include_router(some_router) ，我们需要 default_response_class 在创建期间 设置 APIRouter 。 

from fastapi import APIRouter, FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: str = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter(default_response_class=JSONResponse)


@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):
    """
    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 fastapi import APIRouter, FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: str = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter(default_response_class=JSONResponse)


@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):
    """
    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 ）中 作为查询参数接收的URL， 以及 id 从JSON正文（ 2expen51ve ） 内部发出 的发票 。

添加回调路由器

在这一点上，你有 回调路径操作（S） 需要（一（s）表示， 外部开发者 应该在实现 外部API 在路由器上面创建的回调）。

现在使用的参数 callbacks 在 你的API的路径运行装饰 通过属性 .routes （这其实只是一个 list 路由/ 路径操作 ）从回调路由器： 

from fastapi import APIRouter, FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel, HttpUrl

app = FastAPI()


class Invoice(BaseModel):
    id: str
    title: str = None
    customer: str
    total: float


class InvoiceEvent(BaseModel):
    description: str
    paid: bool


class InvoiceEventReceived(BaseModel):
    ok: bool


invoices_callback_router = APIRouter(default_response_class=JSONResponse)


@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):
    """
    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 。

检查文档

现在，您可以使用Uvicorn启动您的应用程序，然后转到 http://127.0.0.1:8000/docs 。

您将看到文档，其中包括用于 路径操作 的“回调”部分， 该部分显示了 外部API的 外观：

文章来源互联网，如有侵权，请联系管理员删除。邮箱：417803890@qq.com / QQ:417803890

微配音

Python Free

邮箱：417803890@qq.com
QQ：417803890

皖ICP备19001818号
© 2019 copyright www.pythonf.cn - All rights reserved

微信扫一扫关注公众号：

联系方式

Python Free