You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何为SSM参数不存在的场景设置判断条件(Python)

解决SSM Parameter Store批量插入参数的存在性判断问题

你的代码核心问题在于get_parameter方法在参数不存在时不会返回空响应,而是直接抛出ParameterNotFound异常,导致你的判断逻辑无法执行。另外代码里的异常捕获语法也有误(应该是except而非Exception, e)。下面提供两种可行的修正方案:

方案一:单参数查询+异常捕获(适合参数数量较少的场景)

这种方式逐个检查参数,捕获不存在的异常后执行插入:

import json
import boto3
from botocore.exceptions import ClientError

s3 = boto3.resource('s3')
ssm_client = boto3.client('ssm', "us-west-2")

def lambda_handler(event, context):
    bucket = 'bucketjp1'
    key = 'output.json'

    obj = s3.Object(bucket, key)
    data = obj.get()['Body'].read().decode('utf-8')
    json_load = json.loads(data)

    for param in json_load["Parameters"]:
        param_name = param["Name"]
        try:
            # 尝试获取参数,存在则进入else分支
            ssm_client.get_parameter(Name=param_name)
            print(f"The parameter {param_name} already exists.")
        except ClientError as e:
            # 捕获参数不存在的异常,执行插入
            if e.response['Error']['Code'] == 'ParameterNotFound':
                ssm_client.put_parameter(
                    Name=param_name,
                    Type=param["Type"],
                    Value=param["Value"]
                )
                print(f"The parameter {param_name} has been inserted.")
            else:
                # 其他异常正常抛出
                raise e

方案二:批量查询参数列表(适合大量参数的场景)

如果需要插入的参数数量较多,批量查询所有SSM参数名称再做对比,能减少API调用次数,提升效率:

import json
import boto3
from botocore.exceptions import ClientError

s3 = boto3.resource('s3')
ssm_client = boto3.client('ssm', "us-west-2")

def lambda_handler(event, context):
    bucket = 'bucketjp1'
    key = 'output.json'

    obj = s3.Object(bucket, key)
    data = obj.get()['Body'].read().decode('utf-8')
    json_load = json.loads(data)
    params_to_insert = json_load["Parameters"]

    # 批量获取所有SSM参数名称(处理分页)
    existing_param_names = set()
    paginator = ssm_client.get_paginator('describe_parameters')
    for page in paginator.paginate():
        existing_param_names.update(p['Name'] for p in page['Parameters'])

    # 遍历待插入参数,判断是否存在
    for param in params_to_insert:
        param_name = param["Name"]
        if param_name not in existing_param_names:
            ssm_client.put_parameter(
                Name=param_name,
                Type=param["Type"],
                Value=param["Value"]
            )
            print(f"The parameter {param_name} has been inserted.")
        else:
            print(f"The parameter {param_name} already exists.")

关键说明

  • 方案一中必须捕获ClientError并判断错误码ParameterNotFound,因为get_parameter还可能抛出其他类型的异常(如权限不足),需要区分处理。
  • 方案二使用分页器get_paginator获取所有参数,避免因参数数量超过单次API返回上限导致遗漏。
  • 两种方案都避免了原代码中直接调用put_parameter导致的ParameterAlreadyExists错误。

内容的提问来源于stack exchange,提问作者user19882964

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.12 03:31:05