PythonのKeyErrorの解消

Question

前提

AWSで自動電話通知システムを作っています。(Lambda Python3.9)

https://infrastructure-engineer.com/aws-amzaonconnect-008/#index_id3

こちらのサイトを拝見しコードを入力していたのですが下記のようなエラーがでてしまいました。

発生している問題・エラーメッセージ

[ERROR] KeyError: 'cycle'
Traceback (most recent call last):
File "/var/task/lambda_function.py", line 76, in lambda_handler
now_cycle = get_cycle(event)
File "/var/task/lambda_function.py", line 19, in get_cycle
cycle = json_record["cycle"]

該当のソースコード

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

#boto3からDynamoDBへアクセスするためのオブジェクトを取得
dynamodb = boto3.resource('dynamodb')

#"amazonconnect-contact-list"へアクセスするためのオブジェクトを取得
contactlist = dynamodb.Table("amazonconnect-contact-list")

#"amazonconnect-response-status"へアクセスするためのオブジェクトを取得
responsestatus = dynamodb.Table("amazonconnect-response-status")

#残り周回数を取得する関数

def get_cycle(sqs_event):

    record = sqs_event['Records'][0]
    json_record = json.loads(record["body"])
    cycle = json_record["cycle"]

    return cycle


#"amazonconnect-response-status"の内容を返す関数
def status_scan():
    
    scanData = responsestatus.scan()
    items=scanData['Items']

    return scanData["Items"]


#"amazonconnect-contact-list"の内容を返す関数
def operation_scan():

    scanData = contactlist.scan()
    items=scanData['Items']

    return scanData["Items"]


#電話番号リストの行数を返す関数
def get_contactinfo_item_number(json_contactinfo):

    count = 0
    
    for line in json_contactinfo:
        count += 1

    return count


#前回のプライオリティを返す関数
def get_before_priority(sqs_event):

    record = sqs_event['Records'][0]
    json_record = json.loads(record["body"])
    priority = json_record["priority"]

    return priority


#指定されたプライオリティの電話番号を返す関数
def get_phone_number(json_contactinfo, now_priority):
    
    for line in json_contactinfo:
        if line['Priority']==now_priority:
            phone_number = line['Phone']

    return phone_number


def lambda_handler(event, context):

    # 残り周回数を取得
    now_cycle = get_cycle(event)

    # 応答結果を確認
    statustinfo = status_scan()

    # 応答結果がNGの場合
    if statustinfo[0]['CallStatus'] == "NG":
        
        # 電話番号リストを取得
        contactinfo = operation_scan()

        # 電話番号リストの行数を取得
        contactinfo_item_number = get_contactinfo_item_number(contactinfo)

        # 前回のプライオリティを取得
        before_priority = get_before_priority(event)

        # 前回のプラオリティが電話番号リストの行数と同じ場合(リストの最後まで通知した場合)
        if before_priority == contactinfo_item_number:
            
            # プライオリティを1に設定(最初に戻る)
            priority = 1

            # 残り周回数を1減らす
            now_cycle -= 1

        # 前回のプラオリティが電話番号リストの行数と異なる場合(リストの最後まで通知していない場合)
        else:

            # プライオリティに1を足す(次のプライオリティを設定)
            priority = before_priority + 1

        # 残り周回数が0以外の場合
        if now_cycle != 0:

            # 指定したプライオリティの電話番号を取得
            phone_number = get_phone_number(contactinfo, priority)
    
            # boto3からAmazonConnectへアクセスするためのオブジェクトを取得
            connect = boto3.client('connect')
    
            # AmzaonConnectの問い合わせフローを呼び出し電話発信
            connect.start_outbound_voice_contact(
                DestinationPhoneNumber=phone_number,
                ContactFlowId='***',
                InstanceId='***',
                SourcePhoneNumber='***',
            )
    
            # boto3からSQSへアクセスするためのオブジェクトを取得    
            sqs = boto3.resource('sqs')
    
            # "amazonconnect-queue-confirm"へアクセスするためのオブジェクトを取得
            name = 'amazonconnect-queue-confirm'    
            queue = sqs.get_queue_by_name(QueueName=name)
    
            # 現在のプライオリティと周回数をセットしてステータス確認用のキューにメッセージ送信
            response = queue.send_message(MessageBody=json.dumps({"priority": priority, "cycle": now_cycle}))

もし分かる方いらっしゃいましたら回答お願いしたいです。
よろしくお願いします。

※ひとつ前のLambdaも追加させていただきます。
event変数は恐らく 2 でSQSに送信されていると思うのですが…

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

# boto3からDynamoDBへアクセスするためのオブジェクトを取得
dynamodb = boto3.resource('dynamodb')

# "amazonconnect-contact-list"へアクセスするためのオブジェクトを取得
contactlist = dynamodb.Table("amazonconnect-contact-list")

# "amazonconnect-response-status"へアクセスするためのオブジェクトを取得
responsestatus = dynamodb.Table("amazonconnect-response-status")

# CallStatusを"NG"に変更する関数
def status_ng():
    response = responsestatus.update_item(
        Key={
            'No': 1,
            'Name': "Response"
        },
        UpdateExpression="set CallStatus=:c",
        ExpressionAttributeValues={
            ':c': "NG"
        },
        ReturnValues="UPDATED_NEW"
    )
    return response


# "amazonconnect-contact-list"の内容を返す関数
def operation_scan():

    scanData = contactlist.scan()
    items=scanData['Items']

    return scanData["Items"]


# 指定されたプライオリティの電話番号を返す関数
def get_phone_number(json_contactinfo, now_priority):
    
    for line in json_contactinfo:
        if line['Priority']==now_priority:
            phone_number = line['Phone']

    return phone_number


def lambda_handler(event, context):
    
    # 応答結果を初期化
    status_ng()
    
    # 電話番号リストを取得
    contactinfo = operation_scan()

    # 初回のプライオリティを"1"に設定
    priority = 1

    # 周回数を"2"に設定
    cycle = 2

    # 指定したプライオリティの電話番号を取得
    phone_number = get_phone_number(contactinfo, priority)

    # boto3からAmazonConnectへアクセスするためのオブジェクトを取得
    connect = boto3.client('connect')

    # AmzaonConnectの問い合わせフローを呼び出し電話発信
    connect.start_outbound_voice_contact(
        DestinationPhoneNumber=phone_number,
        ContactFlowId='***',
        InstanceId='***',
        SourcePhoneNumber='***',
    )

    # boto3からSQSへアクセスするためのオブジェクトを取得
    sqs = boto3.resource('sqs')

    # "amazonconnect-queue-confirm"へアクセスするためのオブジェクトを取得
    name = 'amazonconnect-queue-confirm'
    queue = sqs.get_queue_by_name(QueueName=name)

    # 現在のプライオリティと周回数をセットしてステータス確認用のキューにメッセージ送信
    response = queue.send_message(MessageBody=json.dumps({"priority": priority, "cycle": cycle}))

Comments

[can110]

もし以下のようなことを追記すると回答得られやすくなると思います。
- event変数の中身およびそれが意図通りか
- 参考にしたサイト（ソースコードの出典）

[konan513]

ありがとうございます。こちらのサイトを参考にしました。
https://infrastructure-engineer.com/aws-amzaonconnect-008/#index_id3

[can110]

この欄は目立たないので質問本文に追記ください。
またソースコードが崩れてインデントがなくなっているのであわせて修正ください。

Answer 1

0. エラーの原因

KeyError: 'cycle'の原因は，辞書型オブジェクトjson_recordが持っていないKeycycleへのアクセスを行ったため，生じたものです．元を辿れば関数lambda_handler(event, context)が受け取ったJSON形式文字列event["Records"][0]["body"]にKeycycleが存在しなかったことに起因します．

したがって，can110さんのコメントで

event変数の中身およびそれが意図通りか

という情報を要求することになっています．

1. event変数の中身が意図通りでないこともある場合

辞書型のKeyアクセスの際に，cycleが存在しないことがある場合でもプログラムを動作させることができます．Keyアクセスに対してデフォルト値を設定することができることを利用します．

def get_cycle(sqs_event):

    record = sqs_event['Records'][0]
    json_record = json.loads(record["body"])
    cycle = json_record.get("cycle", -1) # デフォルト値 -1 を設定しておく

    return cycle

2. event変数の中身が必ず意図通りである場合

辞書型オブジェクトでのアクセス方法が間違っています．中身を教えていただければ適切な手続きを例示できます．

Comments

[konan513]

回答ありがとうございます。初心者すぎてSQSから届いたevent変数の中身を確かめる方法がわからない状態です。。申し訳ありません。
ひとつ前のLambdaも追記します。

[ps_aux_grep]

標準出力に出せば，Log Eventsから見れるようですが，どうでしょうか？
https://dev.classmethod.jp/articles/aws-lambda-json-output/
lambda_handler(event, context)直下でprint(event["Records"][0]["body"])としてみたらどうログが出てきますか？

[konan513]

AWS Lambdaに以下のコードを入力したところ、

import json
def lambda_handler(event, context):
print(event["Records"][0]["body"])

ログに以下のようにでてきました。

{
"priority": 2,
"cycle": 2
}

[ps_aux_grep]

では，そのケースでエラーは起きないはずです．エラーが起きるときはcycleが無かったはずです．