Python SDK使用手冊(cè)
更新時(shí)間:
本文介紹了Python SDK的安裝方法、示例代碼以及如何使用Python SDK發(fā)布事件。
SDK簡(jiǎn)介
事件總線EventBridge的SDK分為管控API SDK和數(shù)據(jù)API SDK,示例代碼和使用的依賴(lài)有所不同。
管控API SDK:對(duì)控制臺(tái)頁(yè)面進(jìn)行操作使用的SDK。
數(shù)據(jù)API SDK:事件數(shù)據(jù)的通道,目前只有發(fā)送事件(PutEvents)屬于此類(lèi)。
前提條件
您已完成以下操作:
環(huán)境準(zhǔn)備
環(huán)境要求
安裝Python 3.4以上版本。更多信息,請(qǐng)參見(jiàn)安裝Python。
說(shuō)明Python 3.4以上版本自帶pip。
查看語(yǔ)言版本
執(zhí)行命令
python -V
查看Python語(yǔ)言版本。
管控API SDK
安裝SDK
執(zhí)行以下命令安裝Python SDK。
pip install alibabacloud_eventbridge20200401==2.0.1
執(zhí)行以下命令安裝打印端口。
pip install alibabacloud_tea_console
SDK示例
您可以在OpenAPI門(mén)戶(hù)通過(guò)調(diào)用API接口,自動(dòng)生成對(duì)應(yīng)的SDK示例。具體操作,請(qǐng)參見(jiàn)如何自動(dòng)生成SDK示例。
數(shù)據(jù)API SDK
安裝SDK
執(zhí)行以下命令安裝Python SDK。
pip install alibabacloud_eventbridge
執(zhí)行以下命令安裝打印端口。
pip install alibabacloud_tea_console
SDK示例
數(shù)據(jù)API SDK目前只包含發(fā)布事件(PutEvents),如需使用Python SDK發(fā)布一個(gè)或多個(gè)事件,請(qǐng)參考以下示例代碼。
# -*- 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初始化公共請(qǐng)求參數(shù)。
"""
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']
# 您的接入點(diǎn)。
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("")
文檔內(nèi)容是否對(duì)您有幫助?