Azure Service Bus Python通用异常处理：如何获取更详细错误信息？

问题 #

开发健康检查功能时，向Service Bus队列发送消息的代码在不同错误场景（如错误访问密钥、防火墙拦截）下，均返回模糊错误提示：

Service Bus has encountered an error. Error condition: amqp:client-error.

使用SDK版本为azure-servicebus==7.11.4，现有代码如下，希望解决两个问题：

1. 如何获取更具描述性的错误信息？
2. 有没有比except Exception as Err:更优的异常处理方式？

现有代码：

def send_message(service_bus_connection_string, queue):
    try:
        # Create Client
        client = ServiceBusClient.from_connection_string(conn_str=service_bus_connection_string)

        # Create Sender.
        sender = client.get_queue_sender(queue_name=queue)

        # Get Date
        date = datetime.now()

        # Create Message
        message_to_send = "Health Check : {}".format(date)
        message = ServiceBusMessage(message_to_send)

        # Send Message
        sender.send_messages(message)

        # Print
        print("Successfully Sent Message:", "[", message, "]")

        # Exit
        client.close()
    except Exception as Err:
        print("Failed to send service bus message, this could be due to an error in the connection string or firewall settings.")
        print(Err)
        exit(1)

解决方案 #

1. 获取更详细的错误信息 #

Azure Service Bus Python SDK的异常实例包含丰富的错误属性，可直接访问获取具体细节：

• error_details：服务端返回的完整错误数据，通常是字典格式，包含message、code等核心字段
• condition：AMQP协议层面的错误标识，辅助定位底层问题
• 配合traceback打印完整调用栈，可排查深层代码执行问题

同时SDK会抛出特定的异常子类，这些子类本身就带有场景化的错误描述，而非笼统的通用异常。

2. 优化异常处理方式 #

不要捕获所有Exception，应针对性捕获Azure Service Bus SDK定义的异常类，既精准处理业务相关错误，又不会意外捕获语法错误等无关异常。

SDK常用的异常类包括：

• ServiceBusAuthenticationError：身份验证失败（如访问密钥错误）
• ServiceBusAuthorizationError：权限不足（如账号无队列操作权限）
• ServiceBusConnectionError：连接失败（如防火墙拦截、网络不通）
• ServiceBusError：所有Service Bus相关错误的基类，可作为兜底捕获

修改后的代码示例 #

from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.exceptions import (
    ServiceBusAuthenticationError,
    ServiceBusAuthorizationError,
    ServiceBusConnectionError,
    ServiceBusError
)
from datetime import datetime
import traceback

def send_message(service_bus_connection_string, queue):
    client = None
    try:
        client = ServiceBusClient.from_connection_string(conn_str=service_bus_connection_string)
        sender = client.get_queue_sender(queue_name=queue)
        
        date = datetime.now()
        message_to_send = "Health Check : {}".format(date)
        message = ServiceBusMessage(message_to_send)
        
        sender.send_messages(message)
        print("Successfully Sent Message:", "[", message, "]")
    except ServiceBusAuthenticationError as auth_err:
        print("身份验证失败：请检查连接字符串中的访问密钥")
        print("详细错误:", auth_err.error_details.get("message", str(auth_err)))
        traceback.print_exc()
        exit(1)
    except ServiceBusAuthorizationError as authz_err:
        print("权限不足：当前账号无操作该队列的权限")
        print("详细错误:", authz_err.error_details.get("message", str(authz_err)))
        traceback.print_exc()
        exit(1)
    except ServiceBusConnectionError as conn_err:
        print("连接失败：请检查网络连接或Service Bus防火墙设置")
        print("详细错误:", conn_err.error_details.get("message", str(conn_err)))
        traceback.print_exc()
        exit(1)
    except ServiceBusError as sb_err:
        print("Service Bus服务错误")
        print("详细错误:", sb_err.error_details.get("message", str(sb_err)))
        print("AMQP错误标识:", sb_err.condition)
        traceback.print_exc()
        exit(1)
    except Exception as unexpected_err:
        # 处理非Service Bus相关的意外错误
        print("发生未知错误")
        print("错误信息:", str(unexpected_err))
        traceback.print_exc()
        exit(1)
    finally:
        # 确保客户端资源被正确释放
        if client:
            client.close()

说明 #

• 捕获特定异常类后，可根据不同错误场景给出针对性提示，无需提前罗列所有潜在问题
• error_details属性能获取服务端返回的原始错误描述，比默认异常字符串更精准
• 使用finally块确保Service Bus客户端被正确关闭，避免资源泄漏
• traceback.print_exc()可打印完整调用栈，方便开发阶段排查问题，生产环境可按需取舍

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