如何在Python中检查DynamoDB属性是否为保留关键字
问题描述
我需要用update_item方法更新DynamoDB表,待更新的属性列表是动态生成的,里面可能包含DynamoDB的保留关键字。有没有办法在Python里自动检查某个属性是不是保留关键字?虽然官方有保留关键字列表,但手动维护静态列表太麻烦。
举个例子:当cols_to_update = ["a", "comment"](comment是保留关键字)时,生成的更新代码如下:
update_expression = "SET a = :_a, comment = :_comment" expression_attribute_values = {":_a": some_value, ":_comment": some_value} table.update_item( Key={ 'id': some_id }, UpdateExpression=update_expression, ExpressionAttributeValues=expression_attribute_values )
执行时会触发报错:
botocore.exceptions.ClientError: An error occurred (ValidationException) when calling the UpdateItem operation: Invalid UpdateExpression: Attribute name is a reserved keyword; reserved keyword: comment
虽然用ExpressionAttributeNames可以解决,但我没法提前知道哪些属性是保留关键字,没法提前配置这个参数。
解决方案
方法1:利用boto3内置的保留关键字检查
boto3依赖的botocore库已经内置了DynamoDB的保留关键字列表,直接调用即可,无需手动维护:
from botocore.docs.dynamodb import reserved_words def is_reserved_word(attr_name): return attr_name.upper() in reserved_words.RESERVED_WORDS # 示例使用 cols_to_update = ["a", "comment"] reserved_attrs = [attr for attr in cols_to_update if is_reserved_word(attr)] print(reserved_attrs) # 输出: ['comment']
方法2:自动生成带别名的更新参数
结合上面的检查逻辑,自动生成合法的UpdateExpression、ExpressionAttributeValues和ExpressionAttributeNames,全程无需手动干预:
from botocore.docs.dynamodb import reserved_words def build_update_params(cols_to_update, value_map): update_expr_parts = [] expr_attr_vals = {} expr_attr_names = {} alias_counter = 0 for col in cols_to_update: val_key = f":_{col}" expr_attr_vals[val_key] = value_map[col] if col.upper() in reserved_words.RESERVED_WORDS: alias = f"#{chr(ord('a') + alias_counter)}" expr_attr_names[alias] = col update_expr_parts.append(f"{alias} = {val_key}") alias_counter += 1 else: update_expr_parts.append(f"{col} = {val_key}") params = { "UpdateExpression": f"SET {', '.join(update_expr_parts)}", "ExpressionAttributeValues": expr_attr_vals } if expr_attr_names: params["ExpressionAttributeNames"] = expr_attr_names return params # 示例使用 cols_to_update = ["a", "comment"] value_map = {"a": "new_a_value", "comment": "new_comment"} update_params = build_update_params(cols_to_update, value_map) # 调用update_item table.update_item( Key={'id': some_id}, **update_params )
方法3:捕获异常动态重试(备选)
如果不想提前检查,也可以捕获ValidationException,从报错信息中提取保留关键字,补上别名后重试:
import botocore def safe_update_item(table, key, update_expr, expr_vals): try: return table.update_item( Key=key, UpdateExpression=update_expr, ExpressionAttributeValues=expr_vals ) except botocore.exceptions.ClientError as e: if e.response['Error']['Code'] == 'ValidationException': error_msg = e.response['Error']['Message'] if 'reserved keyword:' in error_msg: reserved_word = error_msg.split('reserved keyword:')[-1].strip() alias = f"#{reserved_word[0]}" new_update_expr = update_expr.replace(reserved_word, alias) return table.update_item( Key=key, UpdateExpression=new_update_expr, ExpressionAttributeValues=expr_vals, ExpressionAttributeNames={alias: reserved_word} ) # 非保留关键字错误,重新抛出 raise
这种方法适合保留关键字出现频率极低的场景,但需要重试一次,性能略逊于提前检查。
内容的提问来源于stack exchange,提问作者HagaiA
相关产品推荐
相关产品推荐

