如何正确编写Boto3条件判断？DynamoDB查询KeyError报错排查

问题场景 #

创建了包含两条记录的DynamoDB表，其中一条记录的主键是RDS的ARN，另一个字段是名为last_reported的日期字段。运行Lambda函数获取该记录时，预期逻辑是：表中有对应数据则执行后续逻辑，若无数据则退出并提示信息，但实际触发了KeyError:'Item'错误。

错误详情 #

{
    "errorMessage": "'Item'",
    "errorType": "KeyError",
    "requestId": "",
    "stackTrace": [
        "  File \"/var/lang/lib/python3.9/importlib/__init__.py\", line 127, in import_module\n    
        return _bootstrap._gcd_import(name[level:], package, level)\n",
        "  File \"<frozen importlib._bootstrap>\", line 1030, in _gcd_import\n",
        "  File \"<frozen importlib._bootstrap>\", line 1007, in _find_and_load\n",
        "  File \"<frozen importlib._bootstrap>\", line 986, in _find_and_load_unlocked\n",
        "  File \"<frozen importlib._bootstrap>\", line 680, in _load_unlocked\n",
        "  File \"<frozen importlib._bootstrap_external>\", line 850, in exec_module\n",
        "  File \"<frozen importlib._bootstrap>\", line 228, in _call_with_frames_removed\n",
        "  File \"/var/task/lambda_function.py\", line 339, in <module>\n    lambda_handler(1,1)\n",
        "  File \"/var/task/lambda_function.py\", line 263, in lambda_handler\n    last_reported = int(read_db['Item']['last_reported']['N'])\n"
    ]
}

原有问题代码 #

dynamo_table_name = 'my_table'
key = 'arn'
dynamo = boto3.client('dynamodb', region_name=source_region,  config=Config(connect_timeout=5, read_timeout=60, retries={'max_attempts': 10}))
response = dynamo.query(
    TableName=dynamo_table_name, 
    KeyConditionExpression="arn = :arn",
    ExpressionAttributeValues={":arn": {"S":arn}}
)
if response == True:
    read_db = dynamo.get_item(TableName=dynamo_table_name, Key={'arn':{'S':arn}})
    last_reported = int(read_db['Item']['last_reported']['N'])
    print(last_reported)

修正后的代码 #

import boto3
from botocore.config import Config

def lambda_handler(event, context):
    dynamo_table_name = 'my_table'
    arn = "your-rds-arn-here"  # 替换为实际的ARN值
    source_region = "your-region"  # 替换为实际区域
    
    dynamo = boto3.client('dynamodb', region_name=source_region,  
                          config=Config(connect_timeout=5, read_timeout=60, retries={'max_attempts': 10}))
    
    # 直接通过主键获取记录，无需先执行query
    read_db = dynamo.get_item(TableName=dynamo_table_name, Key={'arn': {'S': arn}})
    
    # 检查是否存在目标记录
    if 'Item' in read_db:
        last_reported = int(read_db['Item']['last_reported']['N'])
        print(last_reported)
        # 在这里执行后续业务逻辑
    else:
        print("未找到对应ARN的记录，退出执行")
        return {
            'statusCode': 404,
            'body': "未找到对应记录"
        }

错误原因及修正说明 #

1. 无效的条件判断：原代码中if response == True完全错误，dynamo.query()返回的是包含查询结果的字典，永远不会等于布尔值True，这会导致后续逻辑要么意外执行，要么完全不执行。
2. 未检查记录是否存在：调用dynamo.get_item()后，如果目标主键对应的记录不存在，响应字典中不会包含'Item'键，直接访问read_db['Item']会触发KeyError，必须先判断'Item'是否存在。
3. 冗余的query调用：由于ARN是表的主键，直接使用get_item()即可高效获取单条记录，无需先执行query操作，减少不必要的API调用。

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