如何在DynamoDB中查询最新记录？排查Lambda代码报错

查询DynamoDB中最新录入的记录时，Lambda函数抛出NameError，错误详情如下：

{
  "errorMessage": "name 'Emp_id' is not defined",
  "errorType": "NameError",
  "requestId": "4b66e8ca-28f8-4005-8589-3993d2d4d203",
  "stackTrace": [
    "  File \"/var/task/lambda_function.py\", line 12, in lambda_handler
    hash_key=Emp_id, ScanIndexForward=True, limit=1
"
  ]
}

对应的Lambda函数代码：

import json
import boto3
from boto3.dynamodb.conditions import Key


def lambda_handler(event, context):
 ddb = boto3.resource('dynamodb', region_name='ap-south-1')

 table = ddb.Table('newtable2')

 response = table.query (
        hash_key=Emp_id, ScanIndexForward=True, limit=1
        )
 print(response['Items'])   

1. 解决NameError报错 #

报错核心原因有两个：

• 代码中直接使用了未定义的Emp_id变量；
• table.query()方法的参数格式错误，boto3资源接口的query需要用KeyConditionExpression指定哈希键条件，而非hash_key参数。

2. 实现查询最新记录的逻辑 #

要查询最新录入的记录，前提是表的主键包含排序键（比如存储创建时间戳的created_at、自增ID等），通过倒序排序取第一条即可。

修复后的代码示例（复合主键场景） #

假设表newtable2的哈希键为Emp_id，排序键为created_at（存储记录创建时间），要查询特定员工的最新记录：

import json
import boto3
from boto3.dynamodb.conditions import Key


def lambda_handler(event, context):
    ddb = boto3.resource('dynamodb', region_name='ap-south-1')
    table = ddb.Table('newtable2')
    
    # 替换为实际要查询的员工ID，可从event参数传入或固定值
    target_emp_id = "EMP001"
    
    response = table.query(
        KeyConditionExpression=Key('Emp_id').eq(target_emp_id),
        ScanIndexForward=False,  # 倒序排列，最大的排序键值（最新记录）先返回
        Limit=1  # 只取第一条结果
    )
    
    latest_item = response['Items'][0] if response['Items'] else None
    print(latest_item)
    return {
        'statusCode': 200,
        'body': json.dumps(latest_item)
    }

全表最新记录查询（小表适用） #

如果要查询全表所有记录中的最新一条，小表可以用scan（大表不建议，性能差成本高，推荐创建全局二级索引优化）：

import json
import boto3


def lambda_handler(event, context):
    ddb = boto3.resource('dynamodb', region_name='ap-south-1')
    table = ddb.Table('newtable2')
    
    response = table.scan(
        Limit=1,
        ScanIndexForward=False,  # 按排序键倒序取最新
    )
    
    latest_item = response['Items'][0] if response['Items'] else None
    print(latest_item)
    return {
        'statusCode': 200,
        'body': json.dumps(latest_item)
    }

关键注意事项 #

• 必须确保表有排序键，否则无法高效识别最新记录；
• ScanIndexForward=False是倒序取最新的核心，默认True为正序；
• 生产环境避免用scan查全表最新记录，建议创建全局二级索引，将排序键设为索引的排序键，用query实现高效查询。

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