SDK for Python (Boto3) を使用した Lambda の例 - AWS SDK コードサンプル

Doc AWS SDK Examples リポジトリには、他にも SDK の例があります。 AWS GitHub

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

SDK for Python (Boto3) を使用した Lambda の例

次のコードサンプルは、Lambda で AWS SDK for Python (Boto3) を使用してアクションを実行し、一般的なシナリオを実装する方法を示しています。

アクションはより大きなプログラムからのコードの抜粋であり、コンテキスト内で実行する必要があります。アクションは個々のサービス機能を呼び出す方法を示していますが、関連するシナリオやサービス間の例ではアクションのコンテキストが確認できます。

「シナリオ」は、同じサービス内で複数の関数を呼び出して、特定のタスクを実行する方法を示すコード例です。

各例には、 へのリンクが含まれています。このリンクには GitHub、コンテキスト内でコードをセットアップして実行する方法の手順が記載されています。

アクション

次のコード例は、Lambda 関数を作成する方法を示しています。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def create_function( self, function_name, handler_name, iam_role, deployment_package ): """ Deploys a Lambda function. :param function_name: The name of the Lambda function. :param handler_name: The fully qualified name of the handler function. This must include the file name and the function name. :param iam_role: The IAM role to use for the function. :param deployment_package: The deployment package that contains the function code in .zip format. :return: The Amazon Resource Name (ARN) of the newly created function. """ try: response = self.lambda_client.create_function( FunctionName=function_name, Description="AWS Lambda doc example", Runtime="python3.8", Role=iam_role.arn, Handler=handler_name, Code={"ZipFile": deployment_package}, Publish=True, ) function_arn = response["FunctionArn"] waiter = self.lambda_client.get_waiter("function_active_v2") waiter.wait(FunctionName=function_name) logger.info( "Created function '%s' with ARN: '%s'.", function_name, response["FunctionArn"], ) except ClientError: logger.error("Couldn't create function %s.", function_name) raise else: return function_arn
  • API の詳細については、 CreateFunction AWS SDK for Python (Boto3) API リファレンスの「」を参照してください。

次のコード例は、Lambda 関数を削除する方法を示しています。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def delete_function(self, function_name): """ Deletes a Lambda function. :param function_name: The name of the function to delete. """ try: self.lambda_client.delete_function(FunctionName=function_name) except ClientError: logger.exception("Couldn't delete function %s.", function_name) raise
  • API の詳細については、DeleteFunctionAWS「 SDK for Python (Boto3) API リファレンス」の「」を参照してください。

次のコード例では、Lambda 関数を取得する方法を示します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def get_function(self, function_name): """ Gets data about a Lambda function. :param function_name: The name of the function. :return: The function data. """ response = None try: response = self.lambda_client.get_function(FunctionName=function_name) except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Function %s does not exist.", function_name) else: logger.error( "Couldn't get function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response
  • API の詳細については、 GetFunction AWS SDK for Python (Boto3) API リファレンスの「」を参照してください。

次のコード例は、Lambda 関数を呼び出す方法を示しています。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def invoke_function(self, function_name, function_params, get_log=False): """ Invokes a Lambda function. :param function_name: The name of the function to invoke. :param function_params: The parameters of the function as a dict. This dict is serialized to JSON before it is sent to Lambda. :param get_log: When true, the last 4 KB of the execution log are included in the response. :return: The response from the function invocation. """ try: response = self.lambda_client.invoke( FunctionName=function_name, Payload=json.dumps(function_params), LogType="Tail" if get_log else "None", ) logger.info("Invoked function %s.", function_name) except ClientError: logger.exception("Couldn't invoke function %s.", function_name) raise return response
  • API の詳細については、「AWS SDK for Python (Boto3) API リファレンス」の「[Invoke]」(呼び出し) を参照してください。

以下のコード例では、Lambda 関数を一覧表示する方法を示します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def list_functions(self): """ Lists the Lambda functions for the current account. """ try: func_paginator = self.lambda_client.get_paginator("list_functions") for func_page in func_paginator.paginate(): for func in func_page["Functions"]: print(func["FunctionName"]) desc = func.get("Description") if desc: print(f"\t{desc}") print(f"\t{func['Runtime']}: {func['Handler']}") except ClientError as err: logger.error( "Couldn't list functions. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise
  • API の詳細については、 ListFunctions AWS SDK for Python (Boto3) API リファレンスの「」を参照してください。

次のコード例では、Lambda 関数を更新する方法を示します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def update_function_code(self, function_name, deployment_package): """ Updates the code for a Lambda function by submitting a .zip archive that contains the code for the function. :param function_name: The name of the function to update. :param deployment_package: The function code to update, packaged as bytes in .zip format. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_code( FunctionName=function_name, ZipFile=deployment_package ) except ClientError as err: logger.error( "Couldn't update function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
  • API の詳細については、 UpdateFunctionCode AWS SDK for Python (Boto3) API リファレンスの「」を参照してください。

次のコード例では、Lambda 関数の設定を更新する方法を示します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def update_function_configuration(self, function_name, env_vars): """ Updates the environment variables for a Lambda function. :param function_name: The name of the function to update. :param env_vars: A dict of environment variables to update. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_configuration( FunctionName=function_name, Environment={"Variables": env_vars} ) except ClientError as err: logger.error( "Couldn't update function configuration %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
  • API の詳細については、 UpdateFunctionConfiguration AWS SDK for Python (Boto3) API リファレンスの「」を参照してください。

シナリオ

次のコードサンプルは、以下の操作方法を示しています。

  • IAM ロールと Lambda 関数を作成し、ハンドラーコードをアップロードします。

  • 1 つのパラメーターで関数を呼び出して、結果を取得します。

  • 関数コードを更新し、環境変数で設定します。

  • 新しいパラメーターで関数を呼び出して、結果を取得します。返された実行ログを表示します。

  • アカウントの関数を一覧表示し、リソースをクリーンアップします。

詳細については、「コンソールで Lambda 関数を作成する」を参照してください。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。用例一覧を検索し、AWS コードサンプルリポジトリでの設定と実行の方法を確認してください。

数値をインクリメントする Lambda ハンドラーを定義します。

import logging logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): """ Accepts an action and a single number, performs the specified action on the number, and returns the result. The only allowable action is 'increment'. :param event: The event dict that contains the parameters sent when the function is invoked. :param context: The context in which the function is called. :return: The result of the action. """ result = None action = event.get("action") if action == "increment": result = event.get("number", 0) + 1 logger.info("Calculated result of %s", result) else: logger.error("%s is not a valid action.", action) response = {"result": result} return response

算術演算を実行する 2 番目の Lambda ハンドラーを定義します。

import logging import os logger = logging.getLogger() # Define a list of Python lambda functions that are called by this AWS Lambda function. ACTIONS = { "plus": lambda x, y: x + y, "minus": lambda x, y: x - y, "times": lambda x, y: x * y, "divided-by": lambda x, y: x / y, } def lambda_handler(event, context): """ Accepts an action and two numbers, performs the specified action on the numbers, and returns the result. :param event: The event dict that contains the parameters sent when the function is invoked. :param context: The context in which the function is called. :return: The result of the specified action. """ # Set the log level based on a variable configured in the Lambda environment. logger.setLevel(os.environ.get("LOG_LEVEL", logging.INFO)) logger.debug("Event: %s", event) action = event.get("action") func = ACTIONS.get(action) x = event.get("x") y = event.get("y") result = None try: if func is not None and x is not None and y is not None: result = func(x, y) logger.info("%s %s %s is %s", x, action, y, result) else: logger.error("I can't calculate %s %s %s.", x, action, y) except ZeroDivisionError: logger.warning("I can't divide %s by 0!", x) response = {"result": result} return response

Lambda アクションをラップする関数を作成します。

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource @staticmethod def create_deployment_package(source_file, destination_file): """ Creates a Lambda deployment package in .zip format in an in-memory buffer. This buffer can be passed directly to Lambda when creating the function. :param source_file: The name of the file that contains the Lambda handler function. :param destination_file: The name to give the file when it's deployed to Lambda. :return: The deployment package. """ buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w") as zipped: zipped.write(source_file, destination_file) buffer.seek(0) return buffer.read() def get_iam_role(self, iam_role_name): """ Get an AWS Identity and Access Management (IAM) role. :param iam_role_name: The name of the role to retrieve. :return: The IAM role. """ role = None try: temp_role = self.iam_resource.Role(iam_role_name) temp_role.load() role = temp_role logger.info("Got IAM role %s", role.name) except ClientError as err: if err.response["Error"]["Code"] == "NoSuchEntity": logger.info("IAM role %s does not exist.", iam_role_name) else: logger.error( "Couldn't get IAM role %s. Here's why: %s: %s", iam_role_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return role def create_iam_role_for_lambda(self, iam_role_name): """ Creates an IAM role that grants the Lambda function basic permissions. If a role with the specified name already exists, it is used for the demo. :param iam_role_name: The name of the role to create. :return: The role and a value that indicates whether the role is newly created. """ role = self.get_iam_role(iam_role_name) if role is not None: return role, False lambda_assume_role_policy = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole", } ], } policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole" try: role = self.iam_resource.create_role( RoleName=iam_role_name, AssumeRolePolicyDocument=json.dumps(lambda_assume_role_policy), ) logger.info("Created role %s.", role.name) role.attach_policy(PolicyArn=policy_arn) logger.info("Attached basic execution policy to role %s.", role.name) except ClientError as error: if error.response["Error"]["Code"] == "EntityAlreadyExists": role = self.iam_resource.Role(iam_role_name) logger.warning("The role %s already exists. Using it.", iam_role_name) else: logger.exception( "Couldn't create role %s or attach policy %s.", iam_role_name, policy_arn, ) raise return role, True def get_function(self, function_name): """ Gets data about a Lambda function. :param function_name: The name of the function. :return: The function data. """ response = None try: response = self.lambda_client.get_function(FunctionName=function_name) except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Function %s does not exist.", function_name) else: logger.error( "Couldn't get function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response def create_function( self, function_name, handler_name, iam_role, deployment_package ): """ Deploys a Lambda function. :param function_name: The name of the Lambda function. :param handler_name: The fully qualified name of the handler function. This must include the file name and the function name. :param iam_role: The IAM role to use for the function. :param deployment_package: The deployment package that contains the function code in .zip format. :return: The Amazon Resource Name (ARN) of the newly created function. """ try: response = self.lambda_client.create_function( FunctionName=function_name, Description="AWS Lambda doc example", Runtime="python3.8", Role=iam_role.arn, Handler=handler_name, Code={"ZipFile": deployment_package}, Publish=True, ) function_arn = response["FunctionArn"] waiter = self.lambda_client.get_waiter("function_active_v2") waiter.wait(FunctionName=function_name) logger.info( "Created function '%s' with ARN: '%s'.", function_name, response["FunctionArn"], ) except ClientError: logger.error("Couldn't create function %s.", function_name) raise else: return function_arn def delete_function(self, function_name): """ Deletes a Lambda function. :param function_name: The name of the function to delete. """ try: self.lambda_client.delete_function(FunctionName=function_name) except ClientError: logger.exception("Couldn't delete function %s.", function_name) raise def invoke_function(self, function_name, function_params, get_log=False): """ Invokes a Lambda function. :param function_name: The name of the function to invoke. :param function_params: The parameters of the function as a dict. This dict is serialized to JSON before it is sent to Lambda. :param get_log: When true, the last 4 KB of the execution log are included in the response. :return: The response from the function invocation. """ try: response = self.lambda_client.invoke( FunctionName=function_name, Payload=json.dumps(function_params), LogType="Tail" if get_log else "None", ) logger.info("Invoked function %s.", function_name) except ClientError: logger.exception("Couldn't invoke function %s.", function_name) raise return response def update_function_code(self, function_name, deployment_package): """ Updates the code for a Lambda function by submitting a .zip archive that contains the code for the function. :param function_name: The name of the function to update. :param deployment_package: The function code to update, packaged as bytes in .zip format. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_code( FunctionName=function_name, ZipFile=deployment_package ) except ClientError as err: logger.error( "Couldn't update function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def update_function_configuration(self, function_name, env_vars): """ Updates the environment variables for a Lambda function. :param function_name: The name of the function to update. :param env_vars: A dict of environment variables to update. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_configuration( FunctionName=function_name, Environment={"Variables": env_vars} ) except ClientError as err: logger.error( "Couldn't update function configuration %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def list_functions(self): """ Lists the Lambda functions for the current account. """ try: func_paginator = self.lambda_client.get_paginator("list_functions") for func_page in func_paginator.paginate(): for func in func_page["Functions"]: print(func["FunctionName"]) desc = func.get("Description") if desc: print(f"\t{desc}") print(f"\t{func['Runtime']}: {func['Handler']}") except ClientError as err: logger.error( "Couldn't list functions. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

シナリオを実行する関数を作成します。

class UpdateFunctionWaiter(CustomWaiter): """A custom waiter that waits until a function is successfully updated.""" def __init__(self, client): super().__init__( "UpdateSuccess", "GetFunction", "Configuration.LastUpdateStatus", {"Successful": WaitState.SUCCESS, "Failed": WaitState.FAILURE}, client, ) def wait(self, function_name): self._wait(FunctionName=function_name) def run_scenario(lambda_client, iam_resource, basic_file, calculator_file, lambda_name): """ Runs the scenario. :param lambda_client: A Boto3 Lambda client. :param iam_resource: A Boto3 IAM resource. :param basic_file: The name of the file that contains the basic Lambda handler. :param calculator_file: The name of the file that contains the calculator Lambda handler. :param lambda_name: The name to give resources created for the scenario, such as the IAM role and the Lambda function. """ logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") print("-" * 88) print("Welcome to the AWS Lambda getting started with functions demo.") print("-" * 88) wrapper = LambdaWrapper(lambda_client, iam_resource) print("Checking for IAM role for Lambda...") iam_role, should_wait = wrapper.create_iam_role_for_lambda(lambda_name) if should_wait: logger.info("Giving AWS time to create resources...") wait(10) print(f"Looking for function {lambda_name}...") function = wrapper.get_function(lambda_name) if function is None: print("Zipping the Python script into a deployment package...") deployment_package = wrapper.create_deployment_package( basic_file, f"{lambda_name}.py" ) print(f"...and creating the {lambda_name} Lambda function.") wrapper.create_function( lambda_name, f"{lambda_name}.lambda_handler", iam_role, deployment_package ) else: print(f"Function {lambda_name} already exists.") print("-" * 88) print(f"Let's invoke {lambda_name}. This function increments a number.") action_params = { "action": "increment", "number": q.ask("Give me a number to increment: ", q.is_int), } print(f"Invoking {lambda_name}...") response = wrapper.invoke_function(lambda_name, action_params) print( f"Incrementing {action_params['number']} resulted in " f"{json.load(response['Payload'])}" ) print("-" * 88) print(f"Let's update the function to an arithmetic calculator.") q.ask("Press Enter when you're ready.") print("Creating a new deployment package...") deployment_package = wrapper.create_deployment_package( calculator_file, f"{lambda_name}.py" ) print(f"...and updating the {lambda_name} Lambda function.") update_waiter = UpdateFunctionWaiter(lambda_client) wrapper.update_function_code(lambda_name, deployment_package) update_waiter.wait(lambda_name) print(f"This function uses an environment variable to control logging level.") print(f"Let's set it to DEBUG to get the most logging.") wrapper.update_function_configuration( lambda_name, {"LOG_LEVEL": logging.getLevelName(logging.DEBUG)} ) actions = ["plus", "minus", "times", "divided-by"] want_invoke = True while want_invoke: print(f"Let's invoke {lambda_name}. You can invoke these actions:") for index, action in enumerate(actions): print(f"{index + 1}: {action}") action_params = {} action_index = q.ask( "Enter the number of the action you want to take: ", q.is_int, q.in_range(1, len(actions)), ) action_params["action"] = actions[action_index - 1] print(f"You've chosen to invoke 'x {action_params['action']} y'.") action_params["x"] = q.ask("Enter a value for x: ", q.is_int) action_params["y"] = q.ask("Enter a value for y: ", q.is_int) print(f"Invoking {lambda_name}...") response = wrapper.invoke_function(lambda_name, action_params, True) print( f"Calculating {action_params['x']} {action_params['action']} {action_params['y']} " f"resulted in {json.load(response['Payload'])}" ) q.ask("Press Enter to see the logs from the call.") print(base64.b64decode(response["LogResult"]).decode()) want_invoke = q.ask("That was fun. Shall we do it again? (y/n) ", q.is_yesno) print("-" * 88) if q.ask( "Do you want to list all of the functions in your account? (y/n) ", q.is_yesno ): wrapper.list_functions() print("-" * 88) if q.ask("Ready to delete the function and role? (y/n) ", q.is_yesno): for policy in iam_role.attached_policies.all(): policy.detach_role(RoleName=iam_role.name) iam_role.delete() print(f"Deleted role {lambda_name}.") wrapper.delete_function(lambda_name) print(f"Deleted function {lambda_name}.") print("\nThanks for watching!") print("-" * 88) if __name__ == "__main__": try: run_scenario( boto3.client("lambda"), boto3.resource("iam"), "lambda_handler_basic.py", "lambda_handler_calculator.py", "doc_example_lambda_calculator", ) except Exception: logging.exception("Something went wrong with the demo!")

サーバーレスサンプル

次のコード例では、Kinesis ストリームからレコードを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。この関数は Kinesis ペイロードを取得し、それを Base64 からデコードして、そのレコードの内容をログ記録します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。サーバーレスサンプルリポジトリで完全な例を検索し、設定および実行の方法を確認してください。

Python を使用した Lambda での Kinesis イベントの消費。

import base64 def lambda_handler(event, context): for record in event['Records']: try: print(f"Processed Kinesis Event - EventID: {record['eventID']}") record_data = base64.b64decode(record['kinesis']['data']).decode('utf-8') print(f"Record Data: {record_data}") # TODO: Do interesting work based on the new data except Exception as e: print(f"An error occurred {e}") raise e print(f"Successfully processed {len(event['Records'])} records.")

次のコード例は、S3 バケットにオブジェクトをアップロードすることによってトリガーされるイベントを受け取る Lambda 関数を実装する方法を示しています。この関数は、イベントパラメータから S3 バケット名とオブジェクトキーを取得し、Amazon S3 API を呼び出してオブジェクトのコンテンツタイプを取得してログに記録します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。サーバーレスサンプルリポジトリで完全な例を検索し、設定および実行の方法を確認してください。

Python を使用して Lambda で S3 イベントを消費します。

import json import urllib.parse import boto3 print('Loading function') s3 = boto3.client('s3') def lambda_handler(event, context): #print("Received event: " + json.dumps(event, indent=2)) # Get the object from the event and show its content type bucket = event['Records'][0]['s3']['bucket']['name'] key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8') try: response = s3.get_object(Bucket=bucket, Key=key) print("CONTENT TYPE: " + response['ContentType']) return response['ContentType'] except Exception as e: print(e) print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket)) raise e

次のコード例は、SNS トピックからメッセージを受信することによってトリガーされるイベントを受け取る Lambda 関数を実装する方法を示しています。この関数はイベントパラメータからメッセージを取得し、各メッセージの内容を記録します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。サーバーレスサンプルリポジトリで完全な例を検索し、設定および実行の方法を確認してください。

Python を使用して Lambda で SNS イベントを消費します。

def lambda_handler(event, context): for record in event['Records']: process_message(record) print("done") def process_message(record): try: message = record['Sns']['Message'] print(f"Processed message {message}") # TODO; Process your record here except Exception as e: print("An error occurred") raise e

次のコード例では、SQS キューからメッセージを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。この関数はイベントパラメータからメッセージを取得し、各メッセージの内容を記録します。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。サーバーレスサンプルリポジトリで完全な例を検索し、設定および実行の方法を確認してください。

Python を使用した Lambda での SQS イベントの消費。

def lambda_handler(event, context): for message in event['Records']: process_message(message) print("done") def process_message(message): try: print(f"Processed message {message['body']}") # TODO: Do interesting work based on the new message except Exception as err: print("An error occurred") raise err

以下のコード例では、Kinesis ストリームからイベントを受け取る Lambda 関数のための、部分的なバッチレスポンスの実装方法を示しています。この関数は、レスポンスとしてバッチアイテムの失敗を報告し、対象のメッセージを後で再試行するよう Lambda に伝えます。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。サーバーレスサンプルリポジトリで完全な例を検索し、設定および実行の方法を確認してください。

Python を使用した Lambda での Kinesis バッチアイテム失敗のレポート。

def handler(event, context): records = event.get("Records") curRecordSequenceNumber = "" for record in records: try: # Process your record curRecordSequenceNumber = record["kinesis"]["sequenceNumber"] except Exception as e: # Return failed record's sequence number return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]} return {"batchItemFailures":[]}

以下のコード例では、SQS キューからイベントを受け取る Lambda 関数のための、部分的なバッチレスポンスの実装方法を示しています。この関数は、レスポンスとしてバッチアイテムの失敗を報告し、対象のメッセージを後で再試行するよう Lambda に伝えます。

SDK for Python (Boto3)
注記

には他にもがあります GitHub。サーバーレスサンプルリポジトリで完全な例を検索し、設定および実行の方法を確認してください。

Python を使用した Lambda での SQS バッチアイテム失敗のレポート。

import json def lambda_handler(event, context): if event: batch_item_failures = [] sqs_batch_response = {} for record in event["Records"]: try: # process message except Exception as e: batch_item_failures.append({"itemIdentifier": record['messageId']}) sqs_batch_response["batchItemFailures"] = batch_item_failures return sqs_batch_response