nik/newdwh2021
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: エラー処理を実装（まだ途中）

Browse Source
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2024-02-13 13:04:36 +09:00
parent add64b3bc8
commit 43d959f669
1 changed files with 41 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
lambda/encise-data-unreceive-check/encise-data-unreceive-check.py
View File

@ -122,6 +122,18 @@ def substitute_mail_template(mail_template: str, receive_timing: str, mail_msg:
return mail_str return mail_str
def make_failure_item_on_error(message_id: str) -> dict[str, str]:
"""Report batch item failuresによる処理に失敗したメッセージの判別のためのレスポンスを作成する
@see <https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html#services-sqs-batchfailurereporting>
Args:
message_id (str): SQSメッセージId
Returns:
dict[str, str]: Report batch item failuresで失敗したSQSメッセージを判別するための辞書オブジェクト
"""
return {"itemIdentifier": message_id}
def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, str]], list[str]]: def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, str]], list[str]]:
batch_item_failures = [] batch_item_failures = []
@ -139,11 +151,11 @@ def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, s
logger.info(f'受信したメッセージは既に処理済みのため、処理をスキップします。メッセージID: {message_id}') logger.info(f'受信したメッセージは既に処理済みのため、処理をスキップします。メッセージID: {message_id}')
continue continue
except Exception as e: except Exception as e:
# TODO: エラー処理実装 logger.error(f"E-02-01 メッセージ重複チェック処理に失敗しました　エラー内容：{e}")
raise e batch_item_failures.append(make_failure_item_on_error(message_id))
continue
# SQSパラメータをJSONシリアライズし、Pythonの辞書オブジェクト(イベントパラメータ)を取得する。 # SQSパラメータをJSONシリアライズし、Pythonの辞書オブジェクト(イベントパラメータ)を取得する。
event_parameter = json.loads(record['body']) event_parameter = json.loads(record['body'])
# ③ 設定ファイル[受領チェック対象ファイルリスト]を読み込む # ③ 設定ファイル[受領チェック対象ファイルリスト]を読み込む
@ -158,8 +170,9 @@ def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, s
) )
logger.info('I-03-02 受領チェック対象ファイルリストを読み込みました') logger.info('I-03-02 受領チェック対象ファイルリストを読み込みました')
except Exception as e: except Exception as e:
# TODO: エラー処理実装 logger.error(f"E-03-01 受領チェック対象ファイルリストの読み込みに失敗しました　エラー内容：{e}")
raise e batch_item_failures.append(make_failure_item_on_error(message_id))
continue
# ④ 受領チェック処理を行う # ④ 受領チェック処理を行う
receive_timing = event_parameter['receive_timing'] receive_timing = event_parameter['receive_timing']
@ -232,8 +245,10 @@ def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, s
mail_title_without_line_break = mail_title.splitlines()[0] mail_title_without_line_break = mail_title.splitlines()[0]
logger.info('I-05-04 通知メール（タイトル）テンプレートファイルを読み込みました') logger.info('I-05-04 通知メール（タイトル）テンプレートファイルを読み込みました')
except Exception as e: except Exception as e:
# TODO: エラー処理実装
logger.error(f'E-05-01 通知メール（タイトル）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') logger.error(f'E-05-01 通知メール（タイトル）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}')
raise e batch_item_failures.append(make_failure_item_on_error(message_id))
continue
try: try:
logger.info( logger.info(
@ -250,7 +265,8 @@ def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, s
logger.info('I-05-06 通知メール（本文）テンプレートファイルを読み込みました') logger.info('I-05-06 通知メール（本文）テンプレートファイルを読み込みました')
except Exception as e: except Exception as e:
logger.error(f'E-05-02 通知メール（本文）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}') logger.error(f'E-05-02 通知メール（本文）テンプレートファイルの読み込みに失敗しました　エラー内容：{e}')
raise e batch_item_failures.append(make_failure_item_on_error(message_id))
continue
logger.info(f'I-05-07 メール送信指示をします　送信先トピック：{MBJ_NOTICE_TOPIC}') logger.info(f'I-05-07 メール送信指示をします　送信先トピック：{MBJ_NOTICE_TOPIC}')
params = { params = {
@ -267,27 +283,24 @@ def unreceive_check(records: list, execute_month: str) -> tuple[list[dict[str, s
def lambda_handler(event, context): def lambda_handler(event, context):
try: # ① 処理開始ログを出力する
# ① 処理開始ログを出力する logger.info('I-01-01 処理開始　Enciseデータ受領チェック処理')
logger.info('I-01-01 処理開始　Enciseデータ受領チェック処理') # 処理稼働年月を取得しメモリに保持する
# 処理稼働年月を取得しメモリに保持する execute_date_today = datetime.date.today()
execute_date_today = datetime.date.today() execute_month = execute_date_today.strftime('%Y/%m')
execute_month = execute_date_today.strftime('%Y/%m') logger.info(f'I-01-02 処理稼働月：{execute_month}')
logger.info(f'I-01-02 処理稼働月：{execute_month}') # 処理失敗メッセージIDリストをメモリに保持する（初期値＝空のリスト）
# 処理失敗メッセージIDリストをメモリに保持する（初期値＝空のリスト） batch_item_failures = []
batch_item_failures = [] # 処理成功メッセージIDリストをメモリに保持する（初期値＝空のリスト）
# 処理成功メッセージIDリストをメモリに保持する（初期値＝空のリスト） batch_item_success = []
batch_item_success = [] # ② SQSメッセージ重複排除処理を行う
# ② SQSメッセージ重複排除処理を行う logger.info('I-02-01 メッセージ処理開始')
logger.info('I-02-01 メッセージ処理開始') batch_item_failures, batch_item_success = unreceive_check(event["Records"], execute_month)
batch_item_failures, batch_item_success = unreceive_check(event["Records"], execute_month) logger.info('I-06-01 すべてのメッセージの処理完了')
logger.info('I-06-01 すべてのメッセージの処理完了')
# ⑦ メッセージを処理済として、以下のDynamoDBテーブルに記録する # ⑦ メッセージを処理済として、以下のDynamoDBテーブルに記録する
push_success_messages_to_dynamo_db(batch_item_success) push_success_messages_to_dynamo_db(batch_item_success)
logger.info('I-07-01 処理済みメッセージIDの記録完了') logger.info('I-07-01 処理済みメッセージIDの記録完了')
logger.info('I-07-02 処理終了　Enciseデータ受領チェック処理') logger.info('I-07-02 処理終了　Enciseデータ受領チェック処理')
except Exception as e:
logger.exception(f'E-99 想定外のエラーが発生しました　エラー内容：{e}')
return batch_item_failures return batch_item_failures