使用车队从您的FastAPI API中发布Webhooks-DEV365 开发者社区

webhooks是从应用程序执行操作的应用程序发送的消息（或有效负载）。它们还用于在服务链之间进行交流。例如，付款提供商将Webhook事件发射到操作后的电子商务应用程序。

车队通过用作可靠的出口来促进您的应用程序向客户发布Webhook事件。

ðâ€与传统的Webhook服务器不同，车队允许您重试和重播Webhook事件。

在本文中，您将学习如何使用车队的Python SDK将Webhooks事件发布到FastAPI应用程序中的多个端点。您将首先在Convoy cloud（或Convoy OSS）上设置车队实例，创建一个todo api，最后将车队SDK集成到端点。

。

设置车队

要设置您的车队实例，请登录仪表板并创建一个新项目：

从仪表板创建一个新的传出项目。即将发出的项目是一个用于发送Webhook事件的项目。

将为新创建的传出项目生成项目API密钥。将钥匙存储在安全的地方时，您会在API中使用它来验证您的车队实例以发布Webhooks：

配置您的开发项目。对于本文，您将手动配置您的外向项目。选择*** 没有SDK ***的设置，以手动配置您的传出项目：

ðâ€本文中手动执行的所有动作都可以通过我们的SDKs执行。

创建一个应用程序：应用程序是指带有有效端点的后端应用程序。应用程序可以根据需要包含尽可能多的端点：

每个用户创建了
�IS应用程序，以区分他们的Webhook事件。

创建端点：端点是可以接收Webhook事件的特定目的地。本文中使用的端点URL来自webhooks.site：

为应用程序创建订阅：有关车队将事件连接到其各自应用程序端点的订阅。

您全部设定发布事件并从仪表板监视它们：

在您的车队实例启动和运行时，您将构建API以在下一部分发布您的Webhook事件。

项目设置

从FastAPI驱动的API中发布Webhooks的第一步是构建API。首先创建一个新文件夹并创建虚拟环境：

$ mkdir fastapi-todo-convoy && cd fastapi-todo-convoy
$ python3 -m venv venv

ââsthy您可以使用其他虚拟环境经理，例如Pipenv或poetry。

API结构

概述了API文件夹结构：

fastapi-todo-convoy
 +- .env # store Convoy API key and application ID.
 +- api/
   +- __init__.py # module file
   |
   +- config.py # interact with the environment file
   | 
   +- api.py # todo API source file
   | 
   +- events.py # store for various event types for your webhook payload
 +- main.py # entrypoint to run the application
 +- requirements.txt # application requirements file

在项目文件夹中，添加以下文件：

$ touch .env api/{config,api,events}.py main.py

安装依赖项

作为下一步，激活虚拟环境并安装以下依赖关系：

$ source venv/bin/activate
$ pip install fastapi uvicorn pydantic[dotenv]
$ pip freeze > requirements.txt

最后，安装convoy-python包：

$ pip install git+ssh://git@github.com/frain-dev/convoy-python

设置环境变量

在.env文件中，添加变量：

CONVOY_API_KEY=<your-api-key>
CONVOY_APP_ID=<your-app-id>

用前面生成的API键在上面的环境文件中替换<your-api-key>，并从仪表板中检索车队应用程序ID：

允许您的API检索存储在环境文件中的值，请在config.py中定义BaseSettings子类：

from pydantic import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # convoy credentials
    CONVOY_API_KEY: Optional[str] = None
    CONVOY_APP_ID: Optional[str] = None

    class Config:
        env_file = ".env"

settings = Settings()

上面定义的Settings类读取环境文件中定义的变量。 API可以直接从Settings类settings的实例变量访问变量。

定义事件类型

下一步是定义要在API中发布的Webhook中包含的各种事件有效负载。在[events.py](http://events.py)中，添加代码：

events = {
    "ping": {
        "event": "ping",
        "description": "Webhook test from application."
    },
    "created": {
        "event": "todo.created",
        "description": "Todo created successfully"
    },
    "retrieved": {
        "event": "todo.retrieved",
        "description": "Todo retrieved successfully"
    },
    "updated": {
        "event": "todo.updated",
        "description": "Todo updated successfully"
    },
    "deleted": {
        "event": "todo.deleted",
        "description": "Todo deleted successfully"
    },
    "failed": {
        "event": "todo.failure",
        "description": "Todo not found."
    }
}

在上面的代码块中，您有六种事件类型，所有这些类型都带有todo.，并附有事件的描述。

构建API

API在存储在应用程序内数据库中的todos上执行创建，读取，更新和删除（crud）操作，阵列变量todos。

在api.py中，导入依赖项并创建一个FastAPI的实例：

from convoy import Convoy
from fastapi import FastAPI

from .events import events
from .config import settings

app = FastAPI()

接下来，创建todos变量和车队的实例：

todos = []

convoy = Convoy({"api_key": settings.CONVOY_API_KEY})
app_id = settings.CONVOY_APP_ID

在上面的代码块中，您使用安装的Chevoy-Python SDK使用存储在环境文件中的API键连接到车辆实例。 app_id变量也设置为存储在环境文件中的值。

接下来，您将定义一个函数send_webhook_event，该函数为参数event_type并使用convoy.event.create()方法将事件发布到您的应用程序，并通过扩展为您的应用程序端点。

def send_webhook_event(event_type: str):
    event = {
        "app_id": app_id,
        "event_type": event_type,
        "data": events[event_type]
    }

    (res, err) = convoy.event.create({}, event)
    return res

convoy.event.create()方法采用可选的query和event有效负载主体。事件有效载荷主体是包含应用程序ID，事件类型和有效载荷的对象。在您的API中，事件类型和事件有效载荷已在events.py中定义。

使用Webhook Publisher函数，添加以下代码以完成您的API：

@app.get("/")
async def ping():
    send_webhook_event("ping")
    return {"message": "Wilkomen!"}

@app.get("/todo", tags=["todos"])
async def get_todos() -> dict:
    send_webhook_event("retrieved")
    return { "data": todos }

@app.post("/todo", tags=["todos"])
async def add_todo(todo: dict) -> dict:
    todos.append(todo)
    send_webhook_event("created")
    return {
        "data": { "Todo added." }
    }

@app.put("/todo/{id}", tags=["todos"])
async def update_todo(id: int, body: dict) -> dict:
    for todo in todos:
        if int(todo["id"]) == id:
            todo["item"] = body["item"]
            send_webhook_event("updated")
            return {
                "data": f"Todo with id {id} has been updated."
            }
    send_webhook_event("failed")
    return {
        "data": f"Todo with id {id} not found."
    }

@app.delete("/todo/{id}", tags=["todos"])
async def delete_todo(id: int) -> dict:
    for todo in todos:
        if int(todo["id"]) == id:
            todos.remove(todo)
            send_webhook_event("deleted")
            return {
                "data": f"Todo with id {id} has been removed."
            }

    send_webhook_event("failed")
    return {
        "data": f"Todo with id {id} not found."
    }

您已经定义了在上面的代码块中执行各种CRUD操作的路由。每个成功的操作都会向您的车队应用程序发出Webhook，并且一个失败的操作会发出失败的Webhook事件。

最后，在API的入口点文件main.py中，添加以下内容：

import uvicorn

if __name__ == "__main__":
    uvicorn.run('api.api:app', host="0.0.0.0", port=8080, reload=True)

上面的代码启动了Localhost端口8080上的API。使用命令开始您的应用程序：

$ python main.py

发布网络钩

是时候发布您的第一个Webhook了！要发布您的第一个Webhook事件，请将卷曲请求发送到API：

$ curl http://0.0.0.0:8080/

卷曲请求返回响应：

`{"message":"Wilkomen!"}`

在您的车队仪表板中，活动已交付和记录：

在事件交付选项卡中，您可以验证交付状态以及查看尝试的数量：

如果您单击事件，将显示一个详细的交货页面：

让我们使用curl请求创建一个待办事项，并验证您的API是否会发布Webhook事件：

$ curl -X 'POST' \
  'http://0.0.0.0:8080/todo' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "id": 1,
  "item": "Create a Convoy outgoing project today"
}'

API返回成功的回复：

{
  "data": [
    "Todo added."
  ]
}

create 动作的webhook是否发出？从仪表板验证：

您可以验证API Webhook是否从仪表板成功发射了事件类型created。从事件交付选项卡中可以看到事件交付的详细视图：

访问端点URL从车队仪表板上加强Webhook交付状态：

继续，在TODO API上执行更多操作，并立即在车队上发布Webhooks！

结论

车队提供了在一个请求中将Webhooks发布到各种端点的能力。在本文中，您学会了如何创建和配置车队外传项目，并从构建的API发布到配置的端点。

。

车队为您提供可靠性和可重复性。如果这听起来适合您的建筑，try it out并给我们提供有关Ourslack -Community的反馈！