dynamodb boto3 中的 update_item 示例

Question

在the documentation 之后，我正在尝试创建一个更新语句，如果 dynamodb 表中不存在一个属性，则该语句将更新或添加。

我正在尝试这个

response = table.update_item(
    Key={'ReleaseNumber': '1.0.179'},
    UpdateExpression='SET',
    ConditionExpression='Attr(\'ReleaseNumber\').eq(\'1.0.179\')',
    ExpressionAttributeNames={'attr1': 'val1'},
    ExpressionAttributeValues={'val1': 'false'}
)

我得到的错误是：

botocore.exceptions.ClientError: An error occurred (ValidationException) when calling the UpdateItem operation: ExpressionAttributeNames contains invalid key: Syntax error; key: "attr1"

如果有人做过类似我想要实现的事情，请分享示例。

Answer 1

找到工作示例here，将表的所有索引作为键列出非常重要，这将需要在更新之前进行额外查询，但它可以工作。

response = table.update_item(
    Key={
        'ReleaseNumber': releaseNumber,
        'Timestamp': result[0]['Timestamp']
    },
    UpdateExpression="set Sanity = :r",
    ExpressionAttributeValues={
        ':r': 'false',
    },
    ReturnValues="UPDATED_NEW"
)

Answer 2

使用 boto3 进行 dynamodb 更新的详细信息在网上似乎非常稀少，所以我希望这些替代解决方案有用。

获取/放置

import boto3

table = boto3.resource('dynamodb').Table('my_table')

# get item
response = table.get_item(Key={'pkey': 'asdf12345'})
item = response['Item']

# update
item['status'] = 'complete'

# put (idempotent)
table.put_item(Item=item)

实际更新

import boto3

table = boto3.resource('dynamodb').Table('my_table')

table.update_item(
    Key={'pkey': 'asdf12345'},
    AttributeUpdates={
        'status': 'complete',
    },
)

Answer 3

原始代码示例：

response = table.update_item(
    Key={'ReleaseNumber': '1.0.179'},
    UpdateExpression='SET',
    ConditionExpression='Attr(\'ReleaseNumber\').eq(\'1.0.179\')',
    ExpressionAttributeNames={'attr1': 'val1'},
    ExpressionAttributeValues={'val1': 'false'}
)

固定：

response = table.update_item(
    Key={'ReleaseNumber': '1.0.179'},
    UpdateExpression='SET #attr1 = :val1',
    ConditionExpression=Attr('ReleaseNumber').eq('1.0.179'),
    ExpressionAttributeNames={'#attr1': 'val1'},
    ExpressionAttributeValues={':val1': 'false'}
)

在标记的答案中还显示有一个范围键，因此它也应该包含在Key 中。 update_item 方法必须寻找要更新的确切记录，没有批量更新，并且您不能更新过滤到某个条件的一系列值以获取单个记录。 ConditionExpression 有助于使更新具有幂等性；即如果它已经是那个值，就不要更新它。它不像 sql where 子句。

关于看到的具体错误。

ExpressionAttributeNames 是在 UpdateExpression 中使用的键占位符列表，如果键是保留字，则很有用。

来自文档，“表达式属性名称必须以 # 开头，后跟一个或多个字母数字字符”。错误是因为代码没有使用以# 开头的ExpressionAttributeName，也没有在UpdateExpression 中使用它。

ExpressionAttributeValues 是您要更新到的值的占位符，它们必须以 : 开头

Answer 4

如果您不想逐个检查参数以进行更新，我编写了一个很酷的函数，该函数将返回所需的参数以使用 boto3 执行 update_item 方法。

def get_update_params(body):
    """Given a dictionary we generate an update expression and a dict of values
    to update a dynamodb table.

    Params:
        body (dict): Parameters to use for formatting.

    Returns:
        update expression, dict of values.
    """
    update_expression = ["set "]
    update_values = dict()

    for key, val in body.items():
        update_expression.append(f" {key} = :{key},")
        update_values[f":{key}"] = val

    return "".join(update_expression)[:-1], update_values

这是一个简单的例子：

def update(body):
    a, v = get_update_params(body)
    response = table.update_item(
        Key={'uuid':str(uuid)},
        UpdateExpression=a,
        ExpressionAttributeValues=dict(v)
        )
    return response

Answer 5

基于official example，这是一个简单而完整的解决方案，可用于手动更新（我不推荐）terraform S3 后端使用的表。

假设这是 AWS CLI 显示的表数据：

$ aws dynamodb scan --table-name terraform_lock --region us-east-1
{
    "Items": [
        {
            "Digest": {
                "S": "2f58b12ae16dfb5b037560a217ebd752"
            },
            "LockID": {
                "S": "tf-aws.tfstate-md5"
            }
        }
    ],
    "Count": 1,
    "ScannedCount": 1,
    "ConsumedCapacity": null
}

您可以将其更新为新的摘要（例如您回滚了状态），如下所示：

import boto3

dynamodb = boto3.resource('dynamodb', 'us-east-1')


try:
    table = dynamodb.Table('terraform_lock')
    response = table.update_item(
        Key={
            "LockID": "tf-aws.tfstate-md5"
        },
        UpdateExpression="set Digest=:newDigest",
        ExpressionAttributeValues={
            ":newDigest": "50a488ee9bac09a50340c02b33beb24b"
        },
        ReturnValues="UPDATED_NEW"
    )
except Exception as msg:
    print(f"Oops, could not update: {msg}")

注意":newDigest": "50a488ee9bac09a50340c02b33beb24b" 开头的:，它们很容易错过或忘记。

Answer 6

更新任意数量的dict 属性的示例，并跟踪更新的数量。使用保留字（即name）。

不应使用以下属性名称，因为我们将覆盖值：_inc、_start。

from typing import Dict
from boto3 import Session


def getDynamoDBSession(region: str = "eu-west-1"):
    """Connect to DynamoDB resource from boto3."""
    return Session().resource("dynamodb", region_name=region)


DYNAMODB = getDynamoDBSession()


def updateItemAndCounter(db_table: str, item_key: Dict, attributes: Dict) -> Dict:
    """
    Update item or create new. If the item already exists, return the previous value and
    increase the counter: update_counter.
    """
    table = DYNAMODB.Table(db_table)

    # Init update-expression
    update_expression = "SET"

    # Build expression-attribute-names, expression-attribute-values, and the update-expression
    expression_attribute_names = {}
    expression_attribute_values = {}
    for key, value in attributes.items():
        update_expression += f' #{key} = :{key},'  # Notice the "#" to solve issue with reserved keywords
        expression_attribute_names[f'#{key}'] = key
        expression_attribute_values[f':{key}'] = value

    # Add counter start and increment attributes
    expression_attribute_values[':_start'] = 0
    expression_attribute_values[':_inc'] = 1

    # Finish update-expression with our counter
    update_expression += " update_counter = if_not_exists(update_counter, :_start) + :_inc"

    return table.update_item(
        Key=item_key,
        UpdateExpression=update_expression,
        ExpressionAttributeNames=expression_attribute_names,
        ExpressionAttributeValues=expression_attribute_values,
        ReturnValues="ALL_OLD"
    )

希望它对某人有用！

Answer 7

使用 eltbus 以前的答案，它对我有用，除了小错误，

您必须使用 update_expression[:-1] 删除多余的逗号