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