在 Dynamodb Python 查询中捕获异常答案

【问题标题】：Catching Exception in Dynamodb Python query在 Dynamodb Python 查询中捕获异常
【发布时间】：2020-01-16 17:02:08
【问题描述】：

我正在尝试使用分区键从 dynamodb 获取值，如果找到分区键值，则代码工作正常。但是当没有找到分区键时会抛出异常并且它没有被捕获。

在代码中，我使用了自定义层 d_c_fun，其中定义了记录器功能。我已经尝试过具有以下逻辑的 elseif 函数。如果找到分区键“data_id”，则正确返回值。但是当键值不存在时，它会抛出错误，而不是像下面的异常块一样给出响应。

import json
import boto3
import logging
import d_c_fun as cf

LOGGER = cf.get_logger()
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('tablename')
def lambda_handler(event, context):

    LOGGER.lambda_init(event, context)
    # Event value received through trigger

    D_Id = event['D_Id']
    if not D_Id:
        raise Exception ("data_id value not passed")
        #LOGGER.error(""data_id value not passed"")
    elif D_Id :
        resp = table.get_item(Key={
        "data_id":D_Id})
        LOGGER.info("Response Output value for the %s: %s", D_Id, resp['Item'])
        return resp['Item']
    else:
        raise Exception ("The data_id - %s provided doesn't exist in db", D_Id)

1) o/p：当data_id在下表中匹配时。

输入事件

{ "D_Id": "data2" }

代码输出：

{ "table_name": "data2", "data_id": "data2" }

2) o/p：当data_id不匹配如下表时。

输入事件值：

{ "D_Id": "数据" }

代码输出：

{ "errorMessage": "'Item'", "errorType": "KeyError",
“堆栈跟踪”： [ " File \"/var/task/demo.py\", line 23, in lambda_handler\n LOGGER.info(\"Response Output value for the %s: %s\", D_Id, resp['Item'])\n" ] }

预期的 o/p 应该是在 data_id 不匹配的情况下。

data_id - 提供的数据在 db 中不存在。

3) o/p：当data_id作为空值传递时。

输入事件值：

{ "D_Id": "" }

代码输出：

{ "errorMessage": "data_id 值未通过", "errorType": “异常”，“堆栈跟踪”：[ " File \"/var/task/demo.py\", line 18, in lambda_handler\n raise Exception (\"data_id not set\")\n" ] }

预期的输出应该是。

"errorMessage": "data_id 值未通过",

【问题讨论】：

标签： aws-lambda amazon-dynamodb boto3 dynamodb-queries

【解决方案1】：

我也开发了这段代码，它的工作方式符合预期:)。

    import json
    import boto3
    import logging
    from botocore.exceptions import ClientError

    import d_c_fun as cf

    LOGGER = cf.get_logger()
    def get_dynamodb_client():
    dynamodb = boto3.client('dynamodb', region_name='REGION_NAME')
    return dynamodb


    def get_dynamodb_resource():
    dynamodb = boto3.resource('dynamodb', region_name='REGION_NAME')
    return dynamodb   


    def lambda_handler(event, context):
    D_Id = event['D_Id']

    if not D_Id:
        raise NameError("Field data_id is empty, Please enter data_id")
        return resp['Item']

    resp = get_dynamodb_resource().Table("TABLE_NAME").get_item(Key={
            "data_id":D_Id})

    if 'Item' not in resp:
        raise KeyError(
                'Data "{}" not found. '.format(D_Id) +
                'Please enter valid data_id.'
            )
        return resp['Item']
   else:
       LOGGER.info("Got the Item Successfully - \n %s", resp['Item'])
       return resp['Item']

【讨论】：

这让我发疯了。您的解决方案解决了我的问题。当响应失败时，它没有“项目”，因此您必须使用 if 语句进行检查。我的代码：# Get Employee Name response = table.get_item( Key={'TelephoneNumber': phoneNumber}, ProjectionExpression='EmployeeName' ) if 'Item' not in response: print('Phone number not found: ', phoneNumber) return {'EmployeeName':'0'} else: return response['Item']

【解决方案2】：

我最初的答案完全不符合您的目的。试试这个。

import json
import boto3
import logging
import d_c_fun as cf

LOGGER = cf.get_logger()
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('tablename')
def lambda_handler(event, context):

    LOGGER.lambda_init(event, context)
    # Event value received through trigger

    D_Id = event['D_Id']
    if not D_Id:
        raise Exception ("data_id value not passed")
        #LOGGER.error(""data_id value not passed"")
    else:
        try:
            resp = table.get_item(Key={"data_id":D_Id})
            LOGGER.info("Response Output value for the %s: %s", D_Id, resp['Item'])
            return resp['Item']
        except:
            raise Exception ("The data_id - %s provided doesn't exist in db", D_Id)

【讨论】：