【AWS Lambda関数作成】Python（local環境）の開発 #lambda

はじめに

Python（local環境）の開発に関しての備忘録です。
初心者です😅
間違えてる部分が多々あると思います。
もし見つけた場合、ツッコミいただけると助かります🙇

🦁結論🦁

まずは全体図を作成していくこと

押さえておくべき点

APIGateway は最大30秒でタイムアウトしてしまうため対策が必要。「Lambda発火API」と「Lambda状態取得API」のAPIが必要になる。（このサイトに全てが書かれてる）
「Lambda発火API」は30秒タイムアウトを回避するためにリクエストする
「Lambda状態取得API」に実際の処理をリクエストする
リクエスト、レスポンスは基本json形式
インターバルの設定により、実際の処理経過を確認してる。
インターバルにて’status’を確認してる（RUNNINGが読み込み中、SUCCEEDEDが成功）
ArnはAWSの各機能のアドレス
Local環境に必要なのは「各APIのロールにあるURL」

利用シーン

汎用性の高い処理などをこれから使う場合にAWS Lambdaを使いたいケース

注意点

ループ処理を活用して待機状態を作成してるため、タイムアウトなどの設定がないとアクセスし続けてしまう。
Lambda関数から帰ってくるkeyが同じ名前でなくてはならない（ここでは「lambda_product_name」「lambda_price」）
なるべく、loggerにてログを残す（時間の経過確認、どこの処理に時間がかかっているのかを確認するため）

前提🙊

Csvモジュールについて理解がある。
Jsonモジュールについて理解がある。
Loggerモジュールについて理解がある。
Timeモジュールについて理解がある。
AWSアカウントを持ってる。
StepFunctionsの設定が完了してる。
API Gatewayの設定が完了してる。
Lambda関数が完了してテストも完了してる
各API Gatewayのテストが完了してる。

全体のコード

sample.py

import csv
import json
import requests
import logging
import time

# 最大待機時間と待機間隔（秒）
MAX_WAIT_TIME = 180
POLL_INTERVAL = 10

logging.basicConfig(level=logging.INFO)

goods_status_details = {}


# Lambda状態取得APIの処理を実施
with open('aws_test.csv') as f:
    jans = csv.reader(f)

    for jan in jans:
        json_code = json.dumps({"local_jan_code": jan[0].strip()})  # jsonファイルに置き換え
        print(json_code)

        #  lambda発火APIのURL
        first_api_url = "<ここにlambda発火APIのURLを入力>"
        
        try:
            first_response = requests.post(first_api_url, data=json_code)
            # レスポンス情報をログに記録
            logging.info(f"Lambda発火APIからのレスOK")
        except requests.RequestException as e:
            # エラー情報をログに記録
            logging.error(f"レスエラー: {e}")
            continue



        # Lambda発火APIからのレスポンス処理→レスポンスから実行ARNを抽出
        if first_response.status_code == 200:
            first_response_data = first_response.json()  # 実行Arn取得
            # print("Lambda発火APIからのレスOK")

            execution_arn = first_response_data["executionArn"]

            # Lambda状態取得APIのURL
            second_api_url = "<ここにLambda状態取得APIのURLを入力>"

            second_request_data = {
                "executionArn": execution_arn
            }


            # 開始時間の記録
            start_time = time.time()

            print("lambda関数からのレスポンス待ち", end="")

            # 特定の処理があるまではずっとループ処理する→statusがRUNNINGからSUCCEEDEDになるまで
            while True:
                # Lambda状態取得APIにリクエスト
                second_response = requests.post(second_api_url, json=second_request_data)

                # Lambda状態取得APIからのレスポンス処理
                # 商品名と価格
                if second_response.status_code == 200:
                    second_response_data = second_response.json() # jsonファイルを解析してリストに変換
                    # print(second_response_data)  # レスポンスの詳細

                    # ステータスがSUCCEEDEDになってからの処理　　各値を抽出
                    if second_response_data.get('status') == 'SUCCEEDED':
                        output_dict = second_response_data.get('output')
                        output_data = json.loads(output_dict)
                        body_dict = output_data.get('body')

                        # 各値
                        lambda_product_name = body_dict.get('lambda_product_name')
                        lambda_price = body_dict.get('lambda_price')

                        # 辞書に追加
                        goods_status_details[jan[0]] = (lambda_product_name, lambda_price)

                        print(f"スクレイピング成功:{lambda_product_name},{lambda_price}")
                        break

                    # RUNNINGの際に進捗がわかるように処理
                    elif second_response_data.get('status') == 'RUNNING':
                        current_time = time.time()
                        if current_time - start_time >= 10:  # 10秒ごとにチェック
                            print(".", end="", flush=True)  # flush=Trueは現在の行に '.' を追加
                            start_time = current_time  # 次の10秒間のカウントを開始
                        

                else:
                    # Lambda状態取得APIエラー時の処理
                    print("Lambda状態取得APIエラー:", second_response.text)
                    continue  # 次の反復へ

                if time.time() - start_time > MAX_WAIT_TIME:
                    print("タイムアウト：処理が完了しませんでした。")
                    break

                time.sleep(POLL_INTERVAL)
        else:
            # Lambda発火APIエラー時の処理
            print("Lambda発火APIエラー:", first_response.text)
            continue


# csv出力
with open('output.csv', 'w', newline='') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['JANコード', '商品名', '価格'])
    for jan, (name, price) in goods_status_details.items():
        writer.writerow([jan, name, price])
    print("CSV書き込み完了")

参考にしたサイト📗

APIGatewayタイムアウト不可避を回避（神回避）

AWS Lambda関数作成セクション　リンク