如何在Python中基于CreatedAt属性实现DynamoDB分页与排序？

你遇到的ValidationException错误，原因是DynamoDB的Query操作必须指定哈希键（HASH）的匹配条件——Query是针对主键或全局二级索引（GSI）的分区键进行查询的，你的query_params缺少这个必要参数，因此触发了验证错误。

以下分两种场景给出具体解决方案：

场景1：查询特定Id下的条目，按CreatedAt排序分页 #

如果你的需求是查询某个具体Id对应的所有记录，并按CreatedAt排序分页，只需在query_params中补充哈希键条件，同时通过ScanIndexForward控制排序方向：

修改get_recommended_tales中的query_params #

def get_recommended_tales(event, context):
    limit = int(event['queryStringParameters'].get('limit', 10))
    last_evaluated_key = event['queryStringParameters'].get('lastEvaluatedKey', None)
    # 从请求参数中获取目标Id（根据你的业务逻辑调整来源）
    target_id = event['queryStringParameters'].get('id', '')

    query_params = {
        'TableName': "RecommendedTalesNew",
        'Limit': limit,
        # 指定哈希键匹配条件
        'KeyConditionExpression': 'Id = :id_val',
        'ExpressionAttributeValues': {':id_val': {'S': target_id}},
        # 控制排序方向：False为降序，True为升序（默认）
        'ScanIndexForward': False
    }
    # 后续try-catch逻辑保持不变...

修复分页函数逻辑错误 #

你的get_paged_data函数中，遍历分页器时第一次获取页面就break，会导致只返回第一页数据，即使还有更多内容。优化后的分页函数：

def get_paged_data(paginator_type, query_params, next_token=None):
    dynamodb = client('dynamodb')
    paginator = dynamodb.get_paginator(paginator_type)
    
    if next_token is not None and next_token != "None":
        query_params['ExclusiveStartKey'] = json.loads(next_token)
    
    # 获取分页结果的第一页（已通过Limit控制每页数量）
    page_iterator = paginator.paginate(**query_params)
    first_page = next(page_iterator, {})
    items = first_page.get('Items', [])
    next_token = first_page.get('LastEvaluatedKey', None)
    
    return deseriliaze_dynamodata(items), next_token, next_token is None

场景2：查询所有条目，按CreatedAt全局排序分页 #

如果需要查询表中所有记录并按CreatedAt全局排序，当前表结构无法通过Query高效实现（因为Query必须绑定哈希键，而你的哈希键Id是唯一值）。此时需要调整表结构，新增一个用于全局排序的GSI：

修改SAM表定义，添加全局排序GSI #

RecommendedTalesNewTable:
  Type: 'AWS::DynamoDB::Table'
  Properties:
    TableName: 'RecommendedTalesNew'
    AttributeDefinitions:
      - AttributeName: 'Id'
        AttributeType: 'S'
      - AttributeName: 'CreatedAt'
        AttributeType: 'S'
      # 新增全局分区键属性，所有记录值统一为"ALL"
      - AttributeName: 'GlobalPartitionKey'
        AttributeType: 'S'
    KeySchema:
      - AttributeName: 'Id'
        KeyType: 'HASH'
      - AttributeName: 'CreatedAt'
        KeyType: 'RANGE'
    ProvisionedThroughput:
      ReadCapacityUnits: 5
      WriteCapacityUnits: 5
    GlobalSecondaryIndexes:
      # 保留原GSI（不需要可删除）
      - IndexName: 'CreatedAtIndex'
        KeySchema:
          - AttributeName: 'Id'
            KeyType: 'HASH'
          - AttributeName: 'CreatedAt'
            KeyType: 'RANGE'
        Projection:
          ProjectionType: 'ALL'
        ProvisionedThroughput:
          ReadCapacityUnits: 5
          WriteCapacityUnits: 5
      # 新增全局排序GSI
      - IndexName: 'GlobalCreatedAtIndex'
        KeySchema:
          - AttributeName: 'GlobalPartitionKey'
            KeyType: 'HASH'
          - AttributeName: 'CreatedAt'
            KeyType: 'RANGE'
        Projection:
          ProjectionType: 'ALL'
        ProvisionedThroughput:
          ReadCapacityUnits: 5
          WriteCapacityUnits: 5

注意：写入数据时，需给每条记录添加GlobalPartitionKey字段，值固定为"ALL"。

基于新GSI执行全局排序查询 #

query_params = {
    'TableName': "RecommendedTalesNew",
    # 指定使用新增的全局排序GSI
    'IndexName': 'GlobalCreatedAtIndex',
    'Limit': limit,
    'KeyConditionExpression': 'GlobalPartitionKey = :global_val',
    'ExpressionAttributeValues': {':global_val': {'S': 'ALL'}},
    'ScanIndexForward': False  # 降序排序
}

额外优化建议 #

1. 自动反序列化数据：避免手动编写反序列化函数，使用boto3内置的TypeDeserializer：

from boto3.dynamodb.types import TypeDeserializer

def deseriliaze_dynamodata(items):
    deserializer = TypeDeserializer()
    return [
        {k: deserializer.deserialize(v) for k, v in item.items()}
        for item in items
    ]

2. 使用boto3 Resource简化操作：resource API更简洁，自动处理序列化/反序列化：

import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('RecommendedTalesNew')

# 查询示例
response = table.query(
    KeyConditionExpression=boto3.dynamodb.conditions.Key('Id').eq(target_id),
    Limit=limit,
    ScanIndexForward=False
)

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