调用MicrosoftResourceHealth的emerging_issues.list()遇反序列化错误，如何获取数据？

问题根源 #

你遇到的DeserializationError是因为Azure Resource Health API返回的emerging issues数据中，日期格式用空格分隔日期和时间（比如2023-02-23 11:23:39Z），而Azure Python SDK默认只支持标准ISO 8601格式（用T分隔，比如2023-02-23T11:23:39Z），导致解析失败。其他接口（如availability_statuses.list_by_subscription_id()）返回的日期格式符合标准，所以能正常执行。

解决方案 #

方案1：升级azure-mgmt-resourcehealth SDK版本 #

Azure SDK团队可能已经修复了这种非标准日期格式的解析问题，先尝试升级到最新版本：

pip install --upgrade azure-mgmt-resourcehealth

升级后重新运行代码，大概率能解决问题。

方案2：自定义日期反序列化器（如果升级无效） #

如果升级后问题依然存在，可以自定义一个支持空格分隔日期的反序列化器，替换SDK默认的datetime解析逻辑：

import json
from datetime import datetime
from azure.identity import AzureCliCredential
from azure.mgmt.resourcehealth import MicrosoftResourceHealth
from azure.core.serialization import AzureJSONEncoder
from azure.core.pipeline.policies import SansIOHTTPPolicy

# 自定义日期解析函数，处理空格分隔的格式
def custom_datetime_parser(date_str):
    try:
        # 先尝试标准ISO格式
        return datetime.fromisoformat(date_str.replace('Z', '+00:00'))
    except ValueError:
        # 处理空格分隔的格式
        return datetime.strptime(date_str, '%Y-%m-%d %H:%M:%SZ')

# 自定义反序列化策略
class CustomDeserializerPolicy(SansIOHTTPPolicy):
    def on_response(self, request, response):
        if response.content:
            # 解析响应JSON
            data = json.loads(response.content)
            # 递归处理所有日期字段
            def traverse(obj):
                if isinstance(obj, dict):
                    for k, v in obj.items():
                        if isinstance(v, str):
                            # 尝试识别日期字段（根据emerging issues的字段名调整）
                            if k in ['startTime', 'endTime', 'lastModifiedTime']:
                                try:
                                    obj[k] = custom_datetime_parser(v)
                                except ValueError:
                                    pass
                        elif isinstance(v, (dict, list)):
                            traverse(v)
                elif isinstance(obj, list):
                    for item in obj:
                        traverse(item)
            traverse(data)
            # 替换响应内容为处理后的JSON
            response._content = json.dumps(data, cls=AzureJSONEncoder).encode('utf-8')
        return response

def main():
    resourcehealth_client = MicrosoftResourceHealth(
        credential=AzureCliCredential(),
        subscription_id='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'
    )
    # 添加自定义反序列化策略到客户端管道
    resourcehealth_client._client.pipeline.add_policy(CustomDeserializerPolicy(), after="deserialization")
    
    emerging_events_list = list(resourcehealth_client.emerging_issues.list())
    print(f"There are {len(emerging_events_list)} emerging issues items")

if __name__ == "__main__":
    main()

方案3：直接调用REST API（绕开SDK） #

如果上述两种方案都不适用，可以直接调用Azure Resource Health的REST API，自己处理响应数据：

import requests
from datetime import datetime
from azure.identity import AzureCliCredential

def main():
    credential = AzureCliCredential()
    # 获取访问令牌
    token = credential.get_token("https://management.azure.com/.default").token
    subscription_id = 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'
    
    # 调用emerging issues REST API
    url = f"https://management.azure.com/subscriptions/{subscription_id}/providers/Microsoft.ResourceHealth/emergingIssues?api-version=2022-10-01"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    
    response = requests.get(url, headers=headers)
    response.raise_for_status()
    data = response.json()
    
    # 处理日期字段（按需转换）
    for issue in data.get('value', []):
        # 示例：转换startTime字段
        if 'startTime' in issue:
            try:
                issue['startTime'] = datetime.strptime(issue['startTime'], '%Y-%m-%d %H:%M:%SZ')
            except ValueError:
                pass
    
    print(f"There are {len(data.get('value', []))} emerging issues items")

if __name__ == "__main__":
    main()

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