AWS Secrets Manager创建密钥:默认与用户输入标签合并实现问题
正确实现AWS Secrets Manager标签合并的方法
你的问题核心是数据类型不匹配:AWS Secrets Manager的Tags参数要求传入字典组成的列表,而你之前用{default_tag, tags_dict}生成的是集合,tags_dict, default_tag生成的是元组,自然会触发类型错误。
正确处理逻辑
- 当
tags_dict为None时,直接使用默认标签列表 - 当
tags_dict存在时,合并默认标签和用户标签:可选择两种逻辑,根据需求选用- 逻辑1:简单拼接(若出现重复Key,AWS最终会保留列表中最后一个)
- 逻辑2:用户标签覆盖默认标签(重复Key时优先用户设置的值)
修改后的完整代码
方案1:简单拼接(不处理重复Key)
def create_secret(self, secret_name, secret_value, tags_dict=None): default_tags = [{"Key": "applicationid", "Value": "AP149393"}] # 处理标签逻辑 if tags_dict is None: final_tags = default_tags else: # 校验用户传入的标签格式是否符合要求 if not isinstance(tags_dict, list) or not all(isinstance(t, dict) and "Key" in t and "Value" in t for t in tags_dict): raise ValueError("tags_dict必须是包含{'Key': 'xxx', 'Value': 'xxx'}格式字典的列表") # 合并默认标签与用户标签 final_tags = default_tags + tags_dict create_response = {} for region, client in self.client.items(): try: create_secret = {"Name": secret_name} if isinstance(secret_value, str): create_secret["SecretString"] = secret_value elif isinstance(secret_value, bytes): create_secret["SecretBinary"] = secret_value logging.info(f'Creating secret {secret_name} in the {region} region') # 使用合并后的标签列表 create_response = client.create_secret(**create_secret, Tags=final_tags) except Exception as e: logging.info(f'Could not create a secret {secret_name} in the {region} region. Error: {e}') return create_response
方案2:用户标签覆盖默认标签(处理重复Key)
如果需要避免重复Key冲突,让用户设置的标签覆盖默认值,可先将标签转为字典去重:
def create_secret(self, secret_name, secret_value, tags_dict=None): default_tags = [{"Key": "applicationid", "Value": "AP149393"}] # 先把默认标签转为字典,方便后续合并去重 combined_tag_dict = {tag["Key"]: tag["Value"] for tag in default_tags} if tags_dict is not None: # 校验用户传入的标签格式 if not isinstance(tags_dict, list) or not all(isinstance(t, dict) and "Key" in t and "Value" in t for t in tags_dict): raise ValueError("tags_dict必须是包含{'Key': 'xxx', 'Value': 'xxx'}格式字典的列表") # 合并用户标签,重复Key会覆盖默认值 for tag in tags_dict: combined_tag_dict[tag["Key"]] = tag["Value"] # 转回AWS要求的列表格式 final_tags = [{"Key": key, "Value": val} for key, val in combined_tag_dict.items()] create_response = {} for region, client in self.client.items(): try: create_secret = {"Name": secret_name} if isinstance(secret_value, str): create_secret["SecretString"] = secret_value elif isinstance(secret_value, bytes): create_secret["SecretBinary"] = secret_value logging.info(f'Creating secret {secret_name} in the {region} region') create_response = client.create_secret(**create_secret, Tags=final_tags) except Exception as e: logging.info(f'Could not create a secret {secret_name} in the {region} region. Error: {e}') return create_response
内容的提问来源于stack exchange,提问作者Devrockstar
相关产品推荐
相关产品推荐

