如何使用Boto3查询DynamoDB中存在Email且最近1小时更新的条目

以下是满足需求的完整实现代码，包含FilterExpression的配置：

import boto3
from datetime import datetime, timedelta

# 初始化DynamoDB资源
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)

# 计算1小时前的时间，格式与表中updatedAt字段保持一致
one_hour_ago = (datetime.now() - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S.%f")

# 构建筛选表达式和参数
filter_expression = "attribute_exists(email) AND updatedAt > :one_hour_ago"
expression_attribute_values = {":one_hour_ago": one_hour_ago}

# 执行Scan操作
result = table.scan(
    FilterExpression=filter_expression,
    ExpressionAttributeValues=expression_attribute_values
)

# 处理结果（含分页逻辑，应对单次返回数据量超1MB的情况）
matched_items = result['Items']
while 'LastEvaluatedKey' in result:
    result = table.scan(
        FilterExpression=filter_expression,
        ExpressionAttributeValues=expression_attribute_values,
        ExclusiveStartKey=result['LastEvaluatedKey']
    )
    matched_items.extend(result['Items'])

关键逻辑说明： #

• attribute_exists(email)：精准判断条目是否包含email字段，满足第一个筛选条件。
• updatedAt > :one_hour_ago：将当前时间回溯1小时后，格式化为与表中updatedAt一致的字符串格式，通过比较运算符筛选出最近1小时内更新的数据。
• 分页处理：DynamoDB的Scan单次最多返回1MB数据，通过LastEvaluatedKey可以遍历所有分页数据，确保获取全部符合条件的条目。

内容的提问来源于stack exchange，提问作者ABCD