如何使用事件总线的Python SDK_事件总线(EventBridge)-阿里云帮助中心

本文介绍了Python SDK的安装方法、示例代码以及如何使用Python SDK发布事件。

SDK简介

事件总线EventBridge的SDK分为管控API SDK和数据API SDK，示例代码和使用的依赖有所不同。

管控API SDK：对控制台页面进行操作使用的SDK。
数据API SDK：事件数据的通道，目前只有发送事件（PutEvents）属于此类。

前提条件

您已完成以下操作：

环境准备

环境要求
安装Python 3.4以上版本。更多信息，请参见安装Python。
说明
Python 3.4以上版本自带pip。
查看语言版本
执行命令python -V查看Python语言版本。

管控API SDK

安装SDK

执行以下命令安装Python SDK。

pip install alibabacloud_eventbridge20200401==2.0.1

执行以下命令安装打印端口。
```
pip install alibabacloud_tea_console
```

SDK示例

您可以在OpenAPI门户通过调用API接口，自动生成对应的SDK示例。具体操作，请参见如何自动生成SDK示例。

数据API SDK

安装SDK

执行以下命令安装Python SDK。
```
pip install alibabacloud_eventbridge
```
执行以下命令安装打印端口。
```
pip install alibabacloud_tea_console
```

SDK示例

数据API SDK目前只包含发布事件（PutEvents），如需使用Python SDK发布一个或多个事件，请参考以下示例代码。

# -*- coding: utf-8 -*-

from alibabacloud_eventbridge.client import Client as EventBridgeClient
from alibabacloud_eventbridge import models as event_bridge_models
from alibabacloud_tea_console.client import Client as ConsoleClient
from alibabacloud_tea_util.client import Client as UtilClient


class put_events(object):
    def __init__(self):
        pass

    @staticmethod
    def create_client():
        """
        Create client初始化公共请求参数。
        """
        config = event_bridge_models.Config(

        )
        config.access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
        config.access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        # 您的接入点。
        config.endpoint = "<endpoint>"
        return EventBridgeClient(config)

    @staticmethod
    def put_events(client):
        """
        PutEvents
        """
        event = event_bridge_models.CloudEvent(

        )
        event.datacontenttype = "application/json"
        event.data = UtilClient.to_bytes("test")
        event.id = "a5074581-7e74-4e4c-868f-47e7afdf****"
        event.source = "acs.oss"
        event.specversion = "1.0"
        event.type = "oss:ObjectCreated:PostObject"
        event.time = "2020-08-24T13:54:05.965Asia/Shanghai"
        event.subject = "acs:oss:cn-hangzhou:1234567:xls-papk/game_apk/123.jpg"
        event.extensions = {
            "aliyuneventbusname": "demo-bus"
        }
        try:
            resp = client.put_events([
                event
            ])
            ConsoleClient.log("--------------------Publish event to the aliyun EventBus--------------------")
            ConsoleClient.log(UtilClient.to_jsonstring(resp.to_map()))
        except Exception as error:
            ConsoleClient.log(error.message)

    @staticmethod
    def main(args):
        client = put_events.create_client()
        put_events.put_events(client)

put_events.main("")