dynamodb boto3 中 update_item 的示例
Example of update_item in dynamodb boto3
在 the documentation 之后，我正在尝试创建一个更新语句，如果 dynamodb table.
中不存在只有一个属性，它将更新或添加
我正在尝试这个
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"
如果有人做过与我想要实现的类似的事情，请分享示例。
找到工作示例 here，将 table 的所有索引列为键非常重要，这将需要在更新前进行额外查询，但它有效。
response = table.update_item(
Key={
'ReleaseNumber': releaseNumber,
'Timestamp': result[0]['Timestamp']
},
UpdateExpression="set Sanity = :r",
ExpressionAttributeValues={
':r': 'false',
},
ReturnValues="UPDATED_NEW"
)
关于使用 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',
},
)
原代码示例：
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 中使用的键占位符列表，如果键是保留字则很有用。
来自文档，"An expression attribute name must begin with a #, and be followed by one or more alphanumeric characters"。该错误是因为代码没有使用以 # 开头的 ExpressionAttributeName，也没有在 UpdateExpression.
中使用它
ExpressionAttributeValues 是您要更新到的值的占位符，它们必须以 :
开头
如果您不想逐个参数检查更新，我写了一个很酷的函数，它将 return 执行 update_item[=17 所需的参数=] 方法使用 boto3.
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
基于 official example，这里有一个简单而完整的解决方案，可用于手动更新（我不推荐）由 terraform S3 后端使用的 table。
假设这是 AWS CLI 显示的 table 数据：
$ 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" 开头的 :，它们很容易被遗漏或忘记。
一个示例，用于更新作为 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"
)
希望对某人有用！
使用 eltbus 之前的回答，它对我有用，除了小错误，
您必须使用 update_expression[:-1]
删除多余的逗号
在 the documentation 之后，我正在尝试创建一个更新语句，如果 dynamodb table.
中不存在只有一个属性，它将更新或添加我正在尝试这个
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"
如果有人做过与我想要实现的类似的事情，请分享示例。
找到工作示例 here，将 table 的所有索引列为键非常重要，这将需要在更新前进行额外查询，但它有效。
response = table.update_item(
Key={
'ReleaseNumber': releaseNumber,
'Timestamp': result[0]['Timestamp']
},
UpdateExpression="set Sanity = :r",
ExpressionAttributeValues={
':r': 'false',
},
ReturnValues="UPDATED_NEW"
)
关于使用 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',
},
)
原代码示例：
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 中使用的键占位符列表，如果键是保留字则很有用。
来自文档，"An expression attribute name must begin with a #, and be followed by one or more alphanumeric characters"。该错误是因为代码没有使用以 # 开头的 ExpressionAttributeName，也没有在 UpdateExpression.
ExpressionAttributeValues 是您要更新到的值的占位符，它们必须以 :
如果您不想逐个参数检查更新，我写了一个很酷的函数，它将 return 执行 update_item[=17 所需的参数=] 方法使用 boto3.
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
基于 official example，这里有一个简单而完整的解决方案，可用于手动更新（我不推荐）由 terraform S3 后端使用的 table。
假设这是 AWS CLI 显示的 table 数据：
$ 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" 开头的 :，它们很容易被遗漏或忘记。
一个示例，用于更新作为 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"
)
希望对某人有用！
使用 eltbus 之前的回答，它对我有用，除了小错误，
您必须使用 update_expression[:-1]
删除多余的逗号