如何用Python提取S3存储桶JSON内各对象字段实现动态参数生成

核心逻辑说明 #

你读取到的json_data已经是标准Python列表结构，列表中每个元素对应JSON里的一个设备对象字典，直接遍历或过滤即可提取对应字段。

完整实现代码 #

import boto3
import json

# 原有S3文件读取逻辑
s3 = boto3.resource('s3')
client = boto3.client('s3')
bucket = 'data'
key = 'config.json'
obj = s3.Object(bucket, key)
data = obj.get()['Body'].read().decode('utf-8')
json_data = json.loads(data)

# 遍历所有设备对象生成参数
for device in json_data:
    # 提取字段
    org = device['organisation']
    vendor = device['device_vendor']
    dev_type = device['device_type']
    dev_uuid = device['uuid']
    
    # 生成目标参数
    MQTT_TOPIC = f"{org}/{vendor}/{dev_type}"
    CERTIFICATION_PATH = f"{bucket}/{org}/{dev_uuid}"
    
    # 此处可添加参数的后续使用逻辑，示例打印输出
    print(f"设备UUID：{dev_uuid}")
    print(f"MQTT_TOPIC：{MQTT_TOPIC}")
    print(f"CERTIFICATION_PATH：{CERTIFICATION_PATH}\n")

常用扩展场景 #

提取指定UUID的设备参数 #

如果只需要获取某个特定UUID对应设备的参数，无需遍历全量数据，可以用如下过滤逻辑：

target_uuid = "cde2107e-8736-47de-9e87-2033c3063589"
# 过滤匹配的设备
target_device = next(dev for dev in json_data if dev['uuid'] == target_uuid)

# 生成参数逻辑和上文一致
org = target_device['organisation']
vendor = target_device['device_vendor']
dev_type = target_device['device_type']
MQTT_TOPIC = f"{org}/{vendor}/{dev_type}"
CERTIFICATION_PATH = f"{bucket}/{org}/{target_uuid}"

字段缺失容错处理 #

如果JSON中可能存在字段缺失的情况，可以用get方法替代直接取key，避免抛出KeyError异常：

# 第二个参数为字段缺失时返回的默认值
org = device.get('organisation', 'default_org')
dev_uuid = device.get('uuid', 'default_uuid')

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