如何使用Python查询日期以字符串存储的DynamoDB时间范围数据

核心前提 #

你表中存储的时间格式为YYYY-MM-DD HH:MM:SS.ffffff，属于固定长度补零的时间字符串，字典序和时间先后顺序完全一致，可以直接用字符串大小比较实现范围筛选，无需额外做类型转换，逻辑和你熟悉的SQL里字符串时间范围查询完全一致。

前置检查 #

• 若timestamp字段是表的排序键，或已经配置为全局二级索引（GSI）的排序键：直接用Query接口查询，性能最高、读容量消耗最小，是生产环境推荐方案。
• 若timestamp不是任何索引的键：仅数据量较小的测试场景可以用Scan接口全表扫描加过滤，数据量超过1万条的生产环境必须先给timestamp建GSI，否则会产生极高的读成本和查询延迟。

代码实现 #

首先安装AWS官方Python SDK：

pip install boto3

使用前请提前配置好AWS访问凭证（可通过~/.aws/credentials文件、环境变量配置，和常规AWS工具的凭证配置逻辑一致）。

推荐：Query索引查询（生产环境用） #

import boto3

# 初始化DynamoDB连接，替换为你自己的表所在区域和表名
dynamodb = boto3.resource('dynamodb', region_name='ap-east-1')
table = dynamodb.Table('你的DynamoDB表名')

# 定义查询时间范围，直接用目标时间字符串即可，不需要补微秒
start_ts = '2021-02-17 13:00:00'
end_ts = '2021-02-17 14:00:00'

result_items = []
query_kwargs = {
    # 如果查的是GSI，加一行 IndexName='你的GSI名称',
    # 如果查询需要指定分区键（比如查某用户/某设备下的时间范围），把KeyConditionExpression改成如下格式：
    # 'KeyConditionExpression': '#pk = :pk_val AND #ts BETWEEN :start AND :end',
    # 同时在ExpressionAttributeValues里加 :pk_val 对应的分区键值
    'KeyConditionExpression': '#ts BETWEEN :start AND :end',
    'ExpressionAttributeNames': {
        '#ts': 'timestamp'  # 用占位符避免字段名和DynamoDB保留字冲突
    },
    'ExpressionAttributeValues': {
        ':start': start_ts,
        ':end': end_ts
    }
}

while True:
    resp = table.query(**query_kwargs)
    result_items.extend(resp['Items'])
    # 处理分页：结果超过1MB时DynamoDB会分页返回，循环拉取所有页
    if 'LastEvaluatedKey' not in resp:
        break
    query_kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey']

print(f"共查询到{len(result_items)}条符合条件的数据")

临时方案：Scan全表过滤（仅小表测试用） #

import boto3

dynamodb = boto3.resource('dynamodb', region_name='ap-east-1')
table = dynamodb.Table('你的DynamoDB表名')

start_ts = '2021-02-17 13:00:00'
end_ts = '2021-02-17 14:00:00'

result_items = []
scan_kwargs = {
    'FilterExpression': '#ts BETWEEN :start AND :end',
    'ExpressionAttributeNames': {'#ts': 'timestamp'},
    'ExpressionAttributeValues': {':start': start_ts, ':end': end_ts}
}

while True:
    resp = table.scan(**scan_kwargs)
    result_items.extend(resp['Items'])
    if 'LastEvaluatedKey' not in resp:
        break
    scan_kwargs['ExclusiveStartKey'] = resp['LastEvaluatedKey']

注意事项 #

• 不要把字段名、字段值直接写在条件表达式里，必须用ExpressionAttributeNames和ExpressionAttributeValues传占位符，否则碰到DynamoDB保留字会直接报错。
• 你传入的查询时间不需要补微秒后缀：比如下限2021-02-17 13:00:00会匹配所有大于等于这个值的时间，包括2021-02-17 13:00:00.000000到2021-02-17 13:59:59.999999的所有记录，正好覆盖目标时间范围。
• 必须保证表中存储的时间格式始终是补零的固定长度格式，比如2月必须存02不能存2，否则字符串字典序和时间顺序不一致，会出现查询结果错误。

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