如何让Azure Python函数将所有邮箱地址写入Service Bus队列？

在Azure中创建了一个Service Bus队列，希望开发一个Python版Azure Function，使其能够接收邮箱地址列表并将所有地址写入该队列。

现有代码：

import logging

import azure.functions as func

def main(req: func.HttpRequest, msg: func.Out[str]) -> func.HttpResponse:
    emailaddresses = False

    try:
        req_body = req.get_json()
        emailaddresses = req_body.get('emailaddresses')
    except ValueError:
        pass

    if emailaddresses:
        for address in emailaddresses:
            msg.set(address)
        return func.HttpResponse(f"Successfully processed the following emailaddresses: {', '.join(emailaddresses)}")
    else:
        return func.HttpResponse(
             "No emailaddresses found in body. Pass a list of emailaddresses in the request body.",
             status_code=400
        )

绑定配置：

{ "bindings": [ { "authLevel": "function", "type": "httpTrigger", "direction": "in", "name": "req", "methods": [ "post" ] }, { "type": "http", "direction": "out", "name": "$return" }, { "connection": "servicebusqqqqqqqqqqq_RootManageSharedAccessKey_SERVICEBUS3", "name": "msg", "queueName": "emailaddresses", "direction": "out", "type": "serviceBus" } ] }

测试时函数运行成功，但仅将列表中的最后一个邮箱地址写入Service Bus队列。日志验证循环确实遍历了每个邮箱地址，但每次循环都会覆盖之前的值。了解到C#中可使用ICollector<T>或IAsyncCollector<T>实现多消息写入，但未找到Python的等效实现文档。尝试使用msg: func.Out[typing.List[str]]时出现内部服务器错误，需要调整代码实现需求。

方式一：修改输出绑定实现批量发送 #

问题根源在于func.Out[str]是单条消息输出类型，每次调用set()都会覆盖之前的内容，因此最终仅保留最后一条消息。要发送多条消息，需将输出参数类型改为列表，并一次性传入所有消息的列表。

修改后的代码：

import logging
import typing
import azure.functions as func

def main(req: func.HttpRequest, msg: func.Out[typing.List[str]]) -> func.HttpResponse:
    emailaddresses = None

    try:
        req_body = req.get_json()
        emailaddresses = req_body.get('emailaddresses')
    except ValueError:
        pass

    if emailaddresses and isinstance(emailaddresses, list):
        # 将所有邮箱地址作为多条消息批量发送
        msg.set(emailaddresses)
        return func.HttpResponse(f"Successfully processed the following emailaddresses: {', '.join(emailaddresses)}")
    else:
        return func.HttpResponse(
             "No emailaddresses found in body. Pass a list of emailaddresses in the request body.",
             status_code=400
        )

说明：

• 导入typing模块以支持类型注解
• 将输出参数msg的类型改为func.Out[typing.List[str]]，表示要输出一个字符串列表，每个元素对应一条Service Bus消息
• 移除循环调用msg.set()的逻辑，直接将整个邮箱列表传入msg.set()，运行时会自动将列表中的每个元素作为单独消息发送到队列

如果仍出现内部服务器错误，可尝试将类型注解简化为func.Out[list]，并确保Azure Functions运行时版本为v4及以上。

方式二：使用Service Bus SDK直接发送（更灵活） #

如果输出绑定的方式仍有问题，可直接使用Azure Service Bus Python SDK手动发送多条消息，无需依赖函数绑定：

1. 先安装依赖包：

pip install azure-servicebus

2. 修改后的函数代码：

import logging
import os
import azure.functions as func
from azure.servicebus import ServiceBusClient, ServiceBusMessage

def main(req: func.HttpRequest) -> func.HttpResponse:
    emailaddresses = None
    # 从环境变量获取连接字符串（推荐），或直接填写
    connection_string = os.getenv("servicebusqqqqqqqqqqq_RootManageSharedAccessKey_SERVICEBUS3")
    queue_name = "emailaddresses"

    try:
        req_body = req.get_json()
        emailaddresses = req_body.get('emailaddresses')
    except ValueError:
        pass

    if emailaddresses and isinstance(emailaddresses, list):
        with ServiceBusClient.from_connection_string(connection_string) as client:
            with client.get_queue_sender(queue_name) as sender:
                # 将每个邮箱地址封装为ServiceBusMessage
                messages = [ServiceBusMessage(address) for address in emailaddresses]
                # 批量发送消息
                sender.send_messages(messages)
        return func.HttpResponse(f"Successfully processed the following emailaddresses: {', '.join(emailaddresses)}")
    else:
        return func.HttpResponse(
             "No emailaddresses found in body. Pass a list of emailaddresses in the request body.",
             status_code=400
        )

说明：

• 移除原有的Service Bus输出绑定配置
• 使用ServiceBusClient建立连接，通过send_messages批量发送多条消息
• 推荐从环境变量读取连接字符串，避免硬编码敏感信息

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