基于Policy ID循环调用New Relic API并拼接结果的Python实现求助

需求说明 #

你已经实现了通过分页获取单个告警策略（policy）下NRQL条件的代码，现在需要从给定的JSON字符串中提取所有policy_id，逐个调用API获取对应策略的告警条件，并将所有结果拼接在一起。

完整实现代码 #

import requests
import json

# 请替换为你的实际配置
headers = {"Api-Key": "你的New Relic API密钥"}
proxies = {}  # 有代理需求的话填写代理配置

# 给定的包含告警策略的JSON字符串
policies_json_str = '''
{
  "policies": [
    {
      "id": 1119732,
      "incident_preference": "PER_CONDITION_AND_TARGET",
      "name": "Alert 1",
      "created_at": 1605003072469,
      "updated_at": 1673391349002
    },
    {
      "id": 1133740,
      "incident_preference": "PER_CONDITION",
      "name": "Alert 2",
      "created_at": 1605780951866,
      "updated_at": 1673391348520
    },
    {
      "id": 1136308,
      "incident_preference": "PER_CONDITION",
      "name": "Alert 3",
      "created_at": 1605886254921,
      "updated_at": 1673391349028
    }
  ]
}
'''

def get_single_policy_conditions(policy_id, limit=50):
    """获取单个告警策略下的所有NRQL条件，自动处理分页"""
    policy_conditions = []
    offset = 0
    while True:
        # 构造带policy_id参数的请求URL
        url = f"https://api.newrelic.com/v2/alerts_nrql_conditions.json?policy_id={policy_id}&offset={offset}&limit={limit}"
        response = requests.get(url, headers=headers, proxies=proxies)
        # 检查请求是否成功，失败则抛出异常
        response.raise_for_status()
        data = response.json()
        current_batch = data['data']
        if not current_batch:
            break
        policy_conditions.extend(current_batch)
        offset += limit
    return policy_conditions

# 解析JSON字符串，提取所有policy_id
policies_data = json.loads(policies_json_str)
policy_ids = [policy['id'] for policy in policies_data['policies']]

# 遍历所有policy_id，合并结果
all_conditions = []
for pid in policy_ids:
    print(f"正在获取策略ID {pid} 的告警条件...")
    conditions = get_single_policy_conditions(pid)
    all_conditions.extend(conditions)

# 最终结果存储在all_conditions中
print(f"所有告警条件获取完成，共 {len(all_conditions)} 条记录")

代码说明 #

• JSON解析：用json.loads()将字符串转为字典，通过列表推导式快速提取所有策略ID
• 分页逻辑封装：把原有的分页代码封装成函数，传入策略ID即可获取该策略下所有条件，代码更简洁易复用
• 异常处理：添加response.raise_for_status()，API请求失败时会直接抛出异常，方便排查问题
• 批量处理：遍历每个策略ID，调用函数获取数据并合并到总列表中，最终得到所有策略的告警条件集合

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