本文介紹了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("")
文檔內容是否對您有幫助?