如何用Python创建Azure Service Bus订阅过滤器及管理规则

1. 查看过滤器的具体信息 #

遍历规则时直接打印过滤器对象只会显示内存地址，需通过类型判断和属性访问获取详细信息：

from azure.servicebus.management import ServiceBusAdministrationClient, CorrelationRuleFilter, SqlRuleFilter

CONNECTION_STR = "你的连接字符串"
TOPIC_NAME = "目标主题名"
SUBSCRIPTION_NAME = "目标订阅名"

servicebus_mgmt_client = ServiceBusAdministrationClient.from_connection_string(CONNECTION_STR)

for rule_properties in servicebus_mgmt_client.list_rules(TOPIC_NAME, SUBSCRIPTION_NAME):
    print(f"规则名称: {rule_properties.name}")
    filter_obj = rule_properties.filter
    
    # 获取过滤器类型
    filter_type = type(filter_obj).__name__
    print(f"过滤器类型: {filter_type}")
    
    # 解析CorrelationRuleFilter的详细信息
    if isinstance(filter_obj, CorrelationRuleFilter):
        # 输出内置关联属性
        print(f"关联ID: {filter_obj.correlation_id}")
        print(f"消息ID: {filter_obj.message_id}")
        print(f"目标地址: {filter_obj.to}")
        # 输出自定义属性键值对
        print("自定义属性:")
        for key, value in filter_obj.properties.items():
            print(f"  {key}: {value}")
    
    # 解析SqlRuleFilter的详细信息（如果涉及SQL过滤场景）
    elif isinstance(filter_obj, SqlRuleFilter):
        print(f"SQL过滤表达式: {filter_obj.sql_expression}")
    
    print(f"创建时间: {rule_properties.created_at_utc}")
    print("---")

执行后会输出过滤器的类型（如CorrelationRuleFilter）、内置属性以及所有自定义键值对。

2. 创建规则（含CorrelationRuleFilter） #

create_rule方法需指定规则名称、过滤器实例，可选添加规则动作，正确语法示例如下：

基础用法：创建带CorrelationRuleFilter的规则 #

# 初始化CorrelationRuleFilter，设置内置关联属性
my_filter = CorrelationRuleFilter(
    correlation_id="vip_customer",
    message_id="order_*"
)
# 添加自定义过滤属性
my_filter.properties["customer_level"] = "vip"
my_filter.properties["order_region"] = "north"

# 调用create_rule创建规则
servicebus_mgmt_client.create_rule(
    topic_name=TOPIC_NAME,
    subscription_name=SUBSCRIPTION_NAME,
    rule_name="vip_customer_rule",  # 必须指定规则名称
    filter=my_filter
)

进阶用法：创建带规则动作的规则 #

如果需要在匹配后修改消息属性，可搭配SqlRuleAction：

from azure.servicebus.management import SqlRuleAction

# 定义规则动作：给匹配的消息添加处理标记
my_action = SqlRuleAction(sql_expression="SET IsProcessed = 'True'")

# 创建带动作的规则
servicebus_mgmt_client.create_rule(
    topic_name=TOPIC_NAME,
    subscription_name=SUBSCRIPTION_NAME,
    rule_name="vip_customer_rule_with_action",
    filter=my_filter,
    action=my_action
)

注意：create_rule的rule_name参数为必填项，遗漏会导致调用失败。

内容的提问来源于stack exchange，提问作者Mixie