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

如何在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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.08.10 05:46:08