您可以使用 路径操作 创建一个API,该 操作 可能触发对其 他人(可能与 使用 您的API 的同一开发人员)创建 的 外部API 的请求 。
当您的API应用程序调用 外部API 时发生的过程 称为“回调”。 因为外部开发人员编写的软件会将请求发送到您的API,然后您的API 回叫 ,从而将请求发送到 外部API (可能是同一开发人员创建的)。
在这种情况下,你可能想要记录外部API如何 应该 的样子。 它应该具有 什么 路径操作 ,应该具有什么主体,应该返回什么响应等。
让我们来看一个例子。
想象一下,您开发了一个可以创建发票的应用程序。
这些发票将带有
id
,
title
(可选)
customer
和
total
。
API的用户(外部开发人员)将使用POST请求在您的API中创建发票。
然后,您的API将(让我们想象):
在添加回调之前,让我们首先了解普通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个主要区别 :
pass
。
回调 路径 可以具有 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号-4
© 2019 copyright www.pythonf.cn - All rights reserved
微信扫一扫关注公众号:
Python Free