首页 > 其他分享 >三周精通FastAPI:36 OpenAPI 回调

三周精通FastAPI:36 OpenAPI 回调

时间:2024-11-08 14:20:31浏览次数:3  
标签:invoices FastAPI 36 OpenAPI callback API invoice 回调 外部

官方文档:

OpenAPI 回调

您可以创建触发外部 API 请求的路径操作 API,这个外部 API 可以是别人创建的,也可以是由您自己创建的。

API 应用调用外部 API 时的流程叫做回调。因为外部开发者编写的软件发送请求至您的 API,然后您的 API 要进行回调,并把请求发送至外部 API。

此时,我们需要存档外部 API 的信息,比如应该有哪些路径操作,返回什么样的请求体,应该返回哪种响应等。

使用回调的应用

示例如下。

假设要开发一个创建发票的应用。

发票包括 idtitle(可选)、customertotal 等属性。

API 的用户 (外部开发者)要在您的 API 内使用 POST 请求创建一条发票记录。

(假设)您的 API 将:

  • 把发票发送至外部开发者的消费者
  • 归集现金
  • 把通知发送至 API 的用户(外部开发者)
    • 通过(从您的 API)发送 POST 请求至外部 API (即回调)来完成

常规 FastAPI 应用

添加回调前,首先看下常规 API 应用是什么样子。

常规 API 应用包含接收 Invoice 请求体的路径操作,还有包含回调 URL 的查询参数 callback_url。这部分代码很常规,您对绝大多数代码应该都比较熟悉了:

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/"
requests.post(callback_url, json={"description": "Invoice paid", "paid": True})

但回调最重要的部分可能是,根据 API 要发送给回调请求体的数据等内容,确保您的 API 用户(外部开发者)正确地实现外部 API

因此,我们下一步要做的就是添加代码,为从 API 接收回调的外部 API存档。

这部分文档在 /docs 下的 Swagger API 文档中显示,并且会告诉外部开发者如何构建外部 API

本例没有实现回调本身(只是一行代码),只有文档部分。

"提示"

实际的回调只是 HTTP 请求。

实现回调时,要使用 HTTPX 或 Requests

编写回调文档代码

应用不执行这部分代码,只是用它来记录 外部 API 。

但,您已经知道用 FastAPI 创建自动 API 文档有多简单了。

我们要使用与存档外部 API 相同的知识……通过创建外部 API 要实现的路径操作(您的 API 要调用的)。

"提示"

编写存档回调的代码时,假设您是外部开发者可能会用的上。并且您当前正在实现的是外部 API,不是您自己的 API

临时改变(为外部开发者的)视角能让您更清楚该如何放置外部 API 响应和请求体的参数与 Pydantic 模型等。

创建回调的 APIRouter

首先,新建包含一些用于回调的 APIRouter

from fastapi import APIRouter, FastAPI


invoices_callback_router = APIRouter()

创建回调路径操作

创建回调路径操作也使用之前创建的 APIRouter

它看起来和常规 FastAPI 路径操作差不多:

  • 声明要接收的请求体,例如,body: InvoiceEvent
  • 还要声明要返回的响应,例如,response_model=InvoiceEventReceived

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

回调路径操作与常规路径操作有两点主要区别:

  • 它不需要任何实际的代码,因为应用不会调用这段代码。它只是用于存档外部 API。因此,函数的内容只需要 pass 就可以了
  • 路径可以包含 OpenAPI 3 表达式(详见下文),可以使用带参数的变量,以及发送至您的 API 的原始请求的部分

回调路径表达式

回调路径支持包含发送给您的 API 的原始请求的部分的  OpenAPI 3 表达式

本例中是字符串

"{$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(实际上只是路由/路径操作的列表):

@app.post("/invoices/", callbacks=invoices_callback_router.routes)
def create_invoice(invoice: Invoice, callback_url: Union[HttpUrl, None] = None):

"提示"

注意,不能把路由本身(invoices_callback_router)传递给 callback=,要传递 invoices_callback_router.routes 中的 .routes 属性。

查看文档

现在,使用 Uvicorn 启动应用,打开 http://127.0.0.1:8000/docs。

就能看到文档的路径操作已经包含了回调的内容以及外部 API

实践

源代码

将代码写入callback.py文件

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

启动服务

uvicorn callback:app --reload

测试

curl命令:

curl -X 'POST' \
  'http://127.0.0.1:8000/invoices/?callback_url=http%3A%2F%2Fwww.quye.com' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": "string",
  "title": "string",
  "customer": "string",
  "total": 0
}'

返回信息:

{"msg":"Invoice received"}

证明回调成功。若回调失败,会给出更详细的信息:

{"detail":[{"type":"url_parsing","loc":["query","callback_url"],"msg":"Input should be a valid URL, relative URL without a base","input":"quye.com","ctx":{"error":"relative URL without a base"}}]}

标签:invoices,FastAPI,36,OpenAPI,callback,API,invoice,回调,外部
From: https://blog.csdn.net/skywalk8163/article/details/143484454

相关文章

  • 三周精通FastAPI:37 包含 WSGI - Flask,Django,Pyramid 以及其它
    官方文档:https://fastapi.tiangolo.com/zh/advanced/wsgi/包含WSGI-Flask,Django,其它¶您可以挂载多个WSGI应用,正如您在 SubApplications-Mounts, BehindaProxy 中所看到的那样。为此,您可以使用 WSGIMiddleware 来包装你的WSGI应用,如:Flask,Django,等等。使......
  • CF 1365 题解
    CF1365题解AMatrixGame注意到每次操作都相当于会损失一行和一列,那么最多进行可用行列较少的那一个的轮数.判断奇偶性即可,BTroubleSort手玩发现,不管一个属性的元素集合内部多么无序,都可以借助一个其它属性的元素达到有序.归纳证明特别简单.因此,一个序列可以......
  • 36套Web前端全栈Vue3项目实战P7架构-入门篇+项目篇+进阶篇+架构篇
    36套Web前端全栈Vue3项目实战P7架构-入门篇+项目篇+进阶篇+架构篇36套eb前端全栈Vue3项目实战-入门篇+项目篇+进阶篇+架构篇,P7前端架构,高薪面试,Vue3源码剖析视频课程-技术栈-TypeScript+Vute+ElementPlus+Koa2+Node.js+Pinia+EChart4.0+Uni-App+React18+Flutter+Web3D+Vant+UI,项......
  • CS5366,typec转HDMI,4K60Hz多功能拓展坞方案,低成本替代GSV2201,AG9411方案
    集睿致远ASL新推出的CS5366芯片是一款Type-C转HDMI2Lean4K60的视频转换芯片通过USBType-C连接器将DPRX视频信号转换为HDMI/DVITX视频信号。DP信号转接只用2lane,另外2lane可以输出USB3.0/3.1信号,同时兼容PD3.0,并支持双向PD,相较于CS5266刷新率从30HZ上提升到60HZ,采样频率最......
  • FastAPI 查询参数与字符串校验详解:类型、校验规则与元数据设置
    FastAPI查询参数与字符串校验详解:类型、校验规则与元数据设置本文详细介绍了FastAPI中查询参数的设置与校验方法,涵盖了可选参数、默认值、必要参数和参数列表的处理方式。通过使用Query类,开发者可以为查询参数添加额外的校验规则,如最小长度、最大长度、正则表达式匹配......
  • SCC.369 Working with GPIO Moodle
    SCC.369Coursework1:WorkingwithGPIO Moodlesubmission16:00Fridayweek4;weighting33%ofmodule.AimHavingfamiliarizedyourselfwiththeCdevelopmentenvironmentandthebasicprocessofwritingtoregisterstocontroltheGPIOpins,inthiscour......
  • 如何为管理者设计 360 评估调查题目?
    宣布360评估通常会使管理人员不稳定。同事、下属、管理层甚至客户和供应商通过预先制定的问卷来反馈。360评估可以采用多种形式:从50到300多个问题,例如使用开放式或封闭式问题。但抛开其形式不谈,当360评估与全球人力资源战略保持一致并受其驱动时,它的好处是多方面的。这......
  • 代码随想录算法训练营第十八天|leetcode530.二叉搜索树的最小绝对差、leetcode501.二
    1leetcode530.二叉搜索树的最小绝对差题目链接:530.二叉搜索树的最小绝对差-力扣(LeetCode)文章链接:代码随想录视频链接:你对二叉搜索树了解的还不够!|LeetCode:98.验证二叉搜索树_哔哩哔哩_bilibili思路:定义一个极大值作为结果,然后在中序遍历过程中进行比较出结果1.1自己的......
  • AtCoder Beginner Contest 360 - VP记录
    A-AHealthyBreakfast高桥日常出境。头一次知道getchar()的返回值是int。点击查看代码#include<cstdio>usingnamespacestd;intmain(){ chars[3]={getchar(),getchar(),getchar()}; if(s[0]=='R'&&s[1]=='M')puts("Yes"); els......
  • Last_IO_Error: Got fatal error 1236
    主从机器关机一段时间后,先启动了主库(因为设置了expire_logs_days参数,启动后就会自动删除过期的binlog),启动从库后,发现从库报如下错误,找不到主库的binlog了Last_IO_Error:Gotfatalerror1236frommasterwhenreadingdatafrombinarylog:'Couldnotfindfirstlogf......