EventBridge:EventBridge SDK for Python

Install

pip install alibabacloud_eventbridge20200401==2.0.1
pip install alibabacloud_tea_console

Sample code

OpenAPI Explorer generates ready-to-use Python code for any Management API operation. Select an API operation, configure parameters, and click the Python tab to get the code snippet.

For more information, see Automatic generation of SDK examples.

Install

pip install alibabacloud_eventbridge
pip install alibabacloud_tea_console

Publish events

The Data API SDK supports only the PutEvents operation. The following example publishes a CloudEvent to an event bus named demo-bus.

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

import os
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():
        """
        Initialize the EventBridge client with your credentials and endpoint.
        """
        config = event_bridge_models.Config()
        # Read credentials from environment variables to avoid hardcoding secrets.
        config.access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
        config.access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        # Replace <endpoint> with your EventBridge endpoint.
        # Format: <account-id>.<region-id>.eventbridge.aliyuncs.com
        config.endpoint = "<endpoint>"
        return EventBridgeClient(config)

    @staticmethod
    def put_events(client):
        """
        Build a CloudEvent and publish it to the specified event bus.
        """
        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("")

Replace the following placeholder with your actual value:

Set the following environment variables before running the code:

export ALIBABA_CLOUD_ACCESS_KEY_ID=<your-access-key-id>
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<your-access-key-secret>