如何在Python版Azure函数中配置Service Bus主题触发器的SQL筛选器及应用属性

要实现你需要的功能，核心是在发送Service Bus主题消息时添加自定义应用属性，再给目标订阅配置SQL筛选器匹配这些属性。以下是完整的Python实现方案：

1. 安装依赖 #

首先安装Azure Service Bus的Python SDK及身份验证依赖：

pip install azure-servicebus azure-identity

2. 发送消息时添加应用属性 #

对应你提供的C#代码，Python中通过ServiceBusMessage的application_properties字典来设置自定义属性，这是与C#里ApplicationProperties.Add对应的操作：

import json
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.identity import DefaultAzureCredential

# 替换为你的Service Bus配置信息
SERVICE_BUS_NAMESPACE = "<你的命名空间FQDN，例如xxx.servicebus.windows.net>"
TOPIC_NAME = "<目标主题名称>"

def send_filtered_message(payload):
    # 使用DefaultAzureCredential进行身份验证（生产环境推荐）
    credential = DefaultAzureCredential()
    with ServiceBusClient(SERVICE_BUS_NAMESPACE, credential) as client:
        with client.get_topic_sender(topic_name=TOPIC_NAME) as sender:
            # 序列化消息体
            message_content = json.dumps(payload)
            message = ServiceBusMessage(message_content)
            
            # 添加自定义应用属性，用于订阅筛选
            message.application_properties["goals"] = payload["goals"]
            # 可按需添加多个属性，例如 message.application_properties["priority"] = "high"
            
            # 发送消息
            sender.send_messages(message)
            print(f"已发送消息，goals属性值：{payload['goals']}")

# 测试调用示例
if __name__ == "__main__":
    test_payload = {"id": 101, "content": "季度销售报表", "goals": "sales"}
    send_filtered_message(test_payload)

3. 配置订阅的SQL筛选器 #

通过Azure门户或Azure CLI给目标订阅添加SQL筛选规则，仅让匹配属性的消息进入该订阅：

Azure CLI命令示例 #

az servicebus topic subscription rule create \
    --resource-group <你的资源组名称> \
    --namespace-name <Service Bus命名空间> \
    --topic-name <主题名称> \
    --subscription-name <目标订阅名称> \
    --name "GoalsFilterRule" \
    --filter-sql-expression "goals = 'sales'"

4. Python Azure函数作为订阅触发器 #

当订阅收到符合筛选条件的消息时，会触发你的Azure函数。函数中可读取消息的应用属性：

import azure.functions as func
import json

def main(msg: func.ServiceBusMessage):
    # 解析消息体内容
    msg_body = msg.get_body().decode("utf-8")
    payload = json.loads(msg_body)
    
    # 获取消息的应用属性
    goals_value = msg.application_properties.get("goals")
    print(f"触发函数，匹配的goals属性：{goals_value}")
    print(f"消息内容：{payload}")
    
    # 此处添加你的业务处理逻辑

对应的function.json配置 #

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "msg",
      "type": "serviceBusTrigger",
      "direction": "in",
      "topicName": "<你的主题名称>",
      "subscriptionName": "<你的订阅名称>",
      "connection": "<Service Bus连接字符串的配置项名称>"
    }
  ]
}

内容的提问来源于stack exchange，提问作者Syamala Kumar Rama Koti Reddy