如何用boto3在AWS Secrets Manager追加键值对而不覆盖

AWS Secrets Manager 没有内置的API（包括boto3对应的方法）可以直接追加键值对到现有的秘密中，put_secret_value 本身就是用来替换整个秘密内容的操作。要实现追加键值对的效果，需要自己手动完成「读取现有内容 → 合并新键值 → 重新写入」的流程。

解决方案：扩展你的类实现追加逻辑 #

可以给 SecretsManagerSecret 类添加一个 update_key_value 方法，具体步骤如下：

1. 获取现有秘密的字符串内容
2. 将其解析为Python字典
3. 合并新的键值对（可选择是否覆盖已有相同键）
4. 将合并后的字典转为JSON字符串，重新写入秘密

修改后的完整代码示例：

import boto3
import json
import logging
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)

class SecretsManagerSecret:
    """Encapsulates Secrets Manager functions."""
    def __init__(self, env):
        """
        :param env: Environment on which DAG is running
        """
        self.secretsmanager_client = boto3.client("secretsmanager", region_name="us-east-1")
        self.env = env
        self.name = None

    def put_value(self, secret_value, stages=None):
        if self.name is None:
            raise ValueError("Secret Name/Path not defined")

        try:
            kwargs = {'SecretId': self.name}
            if isinstance(secret_value, str):
                kwargs['SecretString'] = secret_value
            elif isinstance(secret_value, bytes):
                kwargs['SecretBinary'] = secret_value
            if stages is not None:
                kwargs['VersionStages'] = stages
            response = self.secretsmanager_client.put_secret_value(**kwargs)
            logger.info("Value put in secret %s.", self.name)
        except ClientError:
            logger.exception("Couldn't put value in secret %s.", self.name)
            raise
        else:
            return response

    def update_key_value(self, new_key_values, overwrite_existing=True, stages=None):
        """
        追加或更新秘密中的键值对
        :param new_key_values: 要添加的键值对字典，例如 {"new_key": "new_value"}
        :param overwrite_existing: 如果键已存在，是否覆盖原有值，默认True
        :param stages: 版本阶段，可选
        """
        if self.name is None:
            raise ValueError("Secret Name/Path not defined")
        if not isinstance(new_key_values, dict):
            raise TypeError("new_key_values must be a dictionary")

        try:
            # 1. 获取现有秘密内容
            response = self.secretsmanager_client.get_secret_value(SecretId=self.name)
            existing_secret = json.loads(response['SecretString'])
            
            # 2. 合并键值对
            if overwrite_existing:
                existing_secret.update(new_key_values)
            else:
                # 仅添加不存在的键
                for key, value in new_key_values.items():
                    if key not in existing_secret:
                        existing_secret[key] = value
            
            # 3. 写入更新后的内容
            updated_secret_str = json.dumps(existing_secret)
            return self.put_value(updated_secret_str, stages)
        
        except ClientError as e:
            # 处理秘密不存在的情况，如果需要可以自动创建
            if e.response['Error']['Code'] == 'ResourceNotFoundException':
                logger.info("Secret %s not found, creating new one with provided key-values.", self.name)
                return self.put_value(json.dumps(new_key_values), stages)
            logger.exception("Couldn't update secret %s.", self.name)
            raise
        except json.JSONDecodeError:
            logger.exception("Existing secret %s is not valid JSON.", self.name)
            raise

# 使用示例
secret_obj = SecretsManagerSecret("dev")
secret_obj.name = "applications/keys/data_pipelines"

# 追加键值对（不覆盖已有键）
secret_obj.update_key_value({"new_test": "new_value123"}, overwrite_existing=False)

# 或覆盖已有键的更新
secret_obj.update_key_value({"test": "updated_test_value"})

注意事项 #

• 要确保现有秘密的内容是合法的JSON格式，如果是其他格式（比如纯字符串），需要调整解析逻辑
• 操作存在竞态风险：如果在「读取-写入」之间有其他进程修改了秘密，会导致覆盖。如果需要强一致性，可以考虑使用版本ID或添加乐观锁逻辑
• 确保IAM权限足够：需要同时拥有 secretsmanager:GetSecretValue 和 secretsmanager:PutSecretValue 权限

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