You need to enable JavaScript to run this app.
导航
对象存储
  • 文档首页
    • /对象存储/SDK 参考/Python/存储桶接口/管理桶配置/管理事件通知（Python SDK）
管理事件通知（Python SDK）
最近更新时间：2024.02.04 18:31:04首次发布时间：2023.02.02 17:32:41
我的收藏

TOS 支持将部分事件的变动投递到函数服务。如果您希望实时掌握在 TOS 中的资源变动，可以设置事件通知规则。本文介绍事件通知的示例代码。
关于函数服务的详细信息，请参见什么是函数服务。

设置事件通知规则

注意

  • 一个存储桶最多只能创建 10 个事件通知规则。
  • 一个事件通知规则只能选择一个推送目标，即一个函数服务。
  • 一个对象不能匹配多个包含了相同事件类型的规则，即同一个存储桶内，不同规则间类型+前后缀的组合范围不允许重叠。

示例代码

以下代码用于设置桶 bucket-test 的事件通知规则。

import os
import tos
from tos.models2 import CloudFunctionConfiguration, Filter, FilterKey, FilterRule

# 从环境变量获取 AK 和 SK 信息。
ak = os.getenv('TOS_ACCESS_KEY')
sk = os.getenv('TOS_SECRET_KEY')
# your endpoint 和 your region 填写Bucket 所在区域对应的Endpoint。# 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com，your region 填写 cn-beijing。
endpoint = "your endpoint"
region = "your region"
bucket_name = "bucket-test"
cloudFunction = 'your cloud function'
try:
    # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
    client = tos.TosClientV2(ak, sk, endpoint, region)
    config = CloudFunctionConfiguration(
        # 事件通知ID
        id='1',
        # 事件类型
        events=['tos:ObjectCreated:Put', 'tos:ObjectCreated:Post'],
        # 过滤规则
        filter=Filter(
            key=FilterKey(
                rules=[
                    # 设置需要匹配对象的前缀信息
                    FilterRule(name='prefix', value='your prefix'),
                    # 设置需要匹配对象的后缀信息
                    FilterRule(name='suffix', value='your suffix'),
                ]
            )),
        # 函数服务名
        cloud_function=cloudFunction
    )
    out = client.put_bucket_notification(bucket_name, [config])
except tos.exceptions.TosClientError as e:
    # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
    print('fail with client error, message:{}, cause: {}'.format(e.message, e.cause))
except tos.exceptions.TosServerError as e:
    # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
    print('fail with server error, code: {}'.format(e.code))
    # request id 可定位具体问题，强烈建议日志中保存
    print('error with request id: {}'.format(e.request_id))
    print('error with message: {}'.format(e.message))
    print('error with http code: {}'.format(e.status_code))
    print('error with ec: {}'.format(e.ec))
    print('error with request url: {}'.format(e.request_url))
except Exception as e:
    print('fail with unknown error: {}'.format(e))

获取事件通知规则

注意

要获取桶的事件通知规则，默认您必须为桶所有者。

示例代码

以下代码用于获取桶 bucket-test 的事件通知规则。

import os
import tos

# 从环境变量获取 AK 和 SK 信息。
ak = os.getenv('TOS_ACCESS_KEY')
sk = os.getenv('TOS_SECRET_KEY')
# your endpoint 和 your region 填写Bucket 所在区域对应的Endpoint。# 以华北2(北京)为例，your endpoint 填写 tos-cn-beijing.volces.com，your region 填写 cn-beijing。
endpoint = "your endpoint"
region = "your region"
bucket_name = "bucket-test"
try:
    # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
    client = tos.TosClientV2(ak, sk, endpoint, region)
    out = client.get_bucket_notification(bucket_name)
    for n in out.cloud_function_configurations:
        print('id', n.id)
        print('events', n.events)
        print('cloud function', n.cloud_function)
        if n.filter and n.filter.key and n.filter.key.rules:
            for rule in n.filter.key.rules:
                print('filter name', rule.name)
                print('filter value', rule.value)
except tos.exceptions.TosClientError as e:
    # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
    print('fail with client error, message:{}, cause: {}'.format(e.message, e.cause))
except tos.exceptions.TosServerError as e:
    # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
    print('fail with server error, code: {}'.format(e.code))
    # request id 可定位具体问题，强烈建议日志中保存
    print('error with request id: {}'.format(e.request_id))
    print('error with message: {}'.format(e.message))
    print('error with http code: {}'.format(e.status_code))
    print('error with ec: {}'.format(e.ec))
    print('error with request url: {}'.format(e.request_url))
except Exception as e:
    print('fail with unknown error: {}'.format(e))

相关文档

关于事件通知配置的更多信息，请参见事件通知。