Doc AWS SDK Examples GitHub リポジトリには、他にも SDK の例があります。 [AWS](https://github.com/awsdocs/aws-doc-sdk-examples)
翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# SDK for Python (Boto3) を使用した Lambda の例
次のコード例は、Lambda AWS SDK for Python (Boto3) で を使用してアクションを実行し、一般的なシナリオを実装する方法を示しています。
*基本* は、重要なオペレーションをサービス内で実行する方法を示すコード例です。
*アクション*はより大きなプログラムからのコードの抜粋であり、コンテキスト内で実行する必要があります。アクションは個々のサービス機能を呼び出す方法を示していますが、コンテキスト内のアクションは、関連するシナリオで確認できます。
*シナリオ*は、1 つのサービス内から、または他の AWS のサービスと組み合わせて複数の関数を呼び出し、特定のタスクを実行する方法を示すコード例です。
各例には完全なソースコードへのリンクが含まれており、コンテキスト内でコードを設定および実行する方法の手順を確認できます。
**Topics**
+ [はじめに](#get_started)
+ [基本](#basics)
+ [アクション](#actions)
+ [シナリオ](#scenarios)
+ [サーバーレスサンプル](#serverless_examples)
## はじめに
### Hello Lambda
次のコード例は、Lambda の使用を開始する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
import boto3
def main():
"""
List the Lambda functions in your AWS account.
"""
# Create the Lambda client
lambda_client = boto3.client("lambda")
# Use the paginator to list the functions
paginator = lambda_client.get_paginator("list_functions")
response_iterator = paginator.paginate()
print("Here are the Lambda functions in your account:")
for page in response_iterator:
for function in page["Functions"]:
print(f" {function['FunctionName']}")
if __name__ == "__main__":
main()
```
+ API の詳細については、「*AWS SDK for Python (Boto3) の API リファレンス*」の「[ListFunctions](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/ListFunctions)」を参照してください。
## 基本
### 基本を学ぶ
次のコード例は、以下の操作方法を示しています。
+ IAM ロールと Lambda 関数を作成し、ハンドラーコードをアップロードします。
+ 1 つのパラメータで関数を呼び出して、結果を取得します。
+ 関数コードを更新し、環境変数で設定します。
+ 新しいパラメータで関数を呼び出して、結果を取得します。返された実行ログを表示します。
+ アカウントの関数を一覧表示し、リソースをクリーンアップします。
詳細については、「[コンソールで Lambda 関数を作成する](https://docs.aws.amazon.com/lambda/latest/dg/getting-started-create-function.html)」を参照してください。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
数値をインクリメントする 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.9",
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!")
```
+ API の詳細については、「*AWS SDK for Python (Boto3) API リファレンス*」の以下のトピックを参照してください。
+ [CreateFunction](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/CreateFunction)
+ [DeleteFunction](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/DeleteFunction)
+ [GetFunction](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/GetFunction)
+ [Invoke](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/Invoke)
+ [ListFunctions](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/ListFunctions)
+ [UpdateFunctionCode](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/UpdateFunctionCode)
+ [UpdateFunctionConfiguration](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/UpdateFunctionConfiguration)
## アクション
### `CreateFunction`
次のコード例は、`CreateFunction` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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.9",
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 の詳細については、「*AWS SDK for Python (Boto3) の API リファレンス*」の「[CreateFunction](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/CreateFunction)」を参照してください。
### `DeleteFunction`
次のコード例は、`DeleteFunction` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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 の詳細については、「*AWS SDK for Python (Boto3) の API リファレンス*」の「[DeleteFunction](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/DeleteFunction)」を参照してください。
### `GetFunction`
次のコード例は、`GetFunction` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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 の詳細については、*AWS SDK for Python (Boto3) API リファレンス*の「[GetFunction](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/GetFunction)」を参照してください。
### `Invoke`
次のコード例は、`Invoke` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/Invoke)」を参照してください。
### `ListFunctions`
次のコード例は、`ListFunctions` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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 の詳細については、「*AWS SDK for Python (Boto3) の API リファレンス*」の「[ListFunctions](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/ListFunctions)」を参照してください。
### `UpdateFunctionCode`
次のコード例は、`UpdateFunctionCode` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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 の詳細については、「*AWS SDK for Python (Boto3) の API リファレンス*」の「[UpdateFunctionCode](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/UpdateFunctionCode)」を参照してください。
### `UpdateFunctionConfiguration`
次のコード例は、`UpdateFunctionConfiguration` を使用する方法を示しています。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。用例一覧を検索し、[AWS コード例リポジトリ](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#code-examples)での設定と実行の方法を確認してください。
```
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 の詳細については、*AWS SDK for Python (Boto3) API リファレンス*の「[UpdateFunctionConfiguration](https://docs.aws.amazon.com/goto/boto3/lambda-2015-03-31/UpdateFunctionConfiguration)」を参照してください。
## シナリオ
### COVID-19 データを追跡する REST API を作成する
次のコード例は、架空のデータを使用して、米国の COVID-19 の日常的なケースを追跡するシステムをシミュレートする REST API を作成する方法を示しています。
**SDK for Python (Boto3)**
で AWS Chalice を使用して AWS SDK for Python (Boto3) 、Amazon API Gateway AWS Lambda、および Amazon DynamoDB を使用するサーバーレス REST API を作成する方法を示します。REST API は、架空のデータを使用して、米国の COVID-19 の日常的なケースを追跡するシステムをシミュレートします。以下ではその方法を説明しています。
+ AWS Chalice を使用して、API Gateway 経由で送信される REST リクエストを処理するために呼び出される Lambda 関数のルートを定義します。
+ Lambda 関数を使用して、DynamoDB テーブルにデータを取得して保存し、REST リクエストを処理します。
+ AWS CloudFormation テンプレートでテーブル構造とセキュリティロールリソースを定義します。
+ AWS Chalice と CloudFormation を使用して、必要なすべてのリソースをパッケージ化してデプロイします。
+ CloudFormation を使用して、作成されたすべてのリソースをクリーンアップします。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/apigateway_covid-19_tracker) で完全な例を参照してください。
**この例で使用されているサービス**
+ API ゲートウェイ
+ CloudFormation
+ DynamoDB
+ Lambda
### 貸出ライブラリ REST API を作成する
以下のコード例は、Amazon Aurora データベースでバックアップされた REST API を使用して、常連客が書籍の貸出と返却できる貸出ライブラリを作成する方法を示しています。
**SDK for Python (Boto3)**
Amazon Relational Database Service (Amazon RDS) API と AWS Chalice AWS SDK for Python (Boto3) で を使用して、Amazon Aurora データベースにバックアップされた REST API を作成する方法を示します。Web サービスは完全にサーバーレスであり、常連客が本を借りたり返却したりできるシンプルな貸し出しライブラリを表しています。以下ではその方法を説明しています。
+ サーバーレス Aurora データベースクラスターを作成および管理します。
+ AWS Secrets Manager を使用してデータベース認証情報を管理します。
+ Amazon RDS を使用してデータをデータベースに出し入れするデータストレージレイヤーを実装します。
+ AWS Chalice を使用して、サーバーレス REST API を Amazon API Gateway および にデプロイします AWS Lambda。
+ リクエストパッケージを使用して、Web サービスにリクエストを送信します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/aurora_rest_lending_library) で完全な例を参照してください。
**この例で使用されているサービス**
+ API ゲートウェイ
+ Aurora
+ Lambda
+ Secrets Manager
### メッセンジャーアプリケーションを作成する
次のコード例は、データベーステーブルからメッセージレコードを取得する AWS Step Functions メッセンジャーアプリケーションを作成する方法を示しています。
**SDK for Python (Boto3)**
AWS SDK for Python (Boto3) で を使用して AWS Step Functions 、Amazon DynamoDB テーブルからメッセージレコードを取得し、Amazon Simple Queue Service (Amazon SQS) で送信するメッセンジャーアプリケーションを作成する方法を示します。ステートマシンは AWS Lambda 関数と統合して、未送信メッセージがないかデータベースをスキャンします。
+ Amazon DynamoDB テーブルからメッセージレコードを取得および更新するステートマシンを作成します。
+ ステートマシンの定義を更新して、Amazon Simple Queue Service (Amazon SQS) にもメッセージを送信します。
+ ステートマシンの実行を開始および停止します。
+ サービス統合を使用して、ステートマシンから Lambda、DynamoDB、および Amazon SQS に接続します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/stepfunctions_messenger) で完全な例を参照してください。
**この例で使用されているサービス**
+ DynamoDB
+ Lambda
+ Amazon SQS
+ ステップ関数
### WebSocket チャットアプリケーションを作成する
次のコード例は、Amazon API Gateway 上に構築された WebSocket API によって提供されるチャットアプリケーションを作成する方法を示しています。
**SDK for Python (Boto3)**
Amazon API Gateway V2 AWS SDK for Python (Boto3) で を使用して、 AWS Lambda および Amazon DynamoDB と統合する WebSocket API を作成する方法を示します。
+ API Gateway で提供される WebSocket API を作成します。
+ DynamoDB に接続を保存し、他のチャット参加者にメッセージを投稿する Lambda ハンドラを定義します。
+ WebSocket チャットアプリケーションに接続し、WebSockets パッケージを使用してメッセージを送信します。
完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/apigateway_websocket_chat) で完全な例を参照してください。
**この例で使用されているサービス**
+ API ゲートウェイ
+ DynamoDB
+ Lambda
### API Gateway を使用して Lambda 関数を呼び出す
次のコード例は、Amazon API Gateway によって呼び出される AWS Lambda 関数を作成する方法を示しています。
**SDK for Python (Boto3)**
この例は、 AWS Lambda 関数を対象とする Amazon API Gateway REST API を作成して使用する方法を示しています。Lambda ハンドラーは、HTTP メソッドに基づいてルーティングする方法を示します。クエリ文字列、ヘッダー、および本文からデータを取得する方法。そして、JSON 応答を返す方法。
+ Lambda 関数をデプロイします。
+ API ゲートウェイ REST API を作成します。
+ Lambda 関数をターゲットとする REST リソースを作成します。
+ API Gateway に Lambda 関数を呼び出す権限を付与します。
+ リクエストパッケージを使用して、REST API にリクエストを送信します。
+ デモ中に作成されたすべてのリソースをクリーンアップします。
この例は GitHub で最もよく確認できます。完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#readme) で完全な例を参照してください。
**この例で使用されているサービス**
+ API ゲートウェイ
+ DynamoDB
+ Lambda
+ Amazon SNS
### スケジュールされたイベントを使用した Lambda 関数の呼び出し
次のコード例は、Amazon EventBridge スケジュールされたイベントによって呼び出される AWS Lambda 関数を作成する方法を示しています。
**SDK for Python (Boto3)**
この例では、スケジュールされた Amazon EventBridge イベントのターゲットとして AWS Lambda 関数を登録する方法を示します。Lambda ハンドラーは、後で取得するために Amazon CloudWatch Logs にわかりやすいメッセージと完全なイベントデータを書き込みます。
+ Lambda 関数をデプロイします。
+ EventBridge スケジュールイベントを作成し、Lambda 関数をターゲットにします。
+ EventBridge に Lambda 関数を呼び出す許可を付与します
+ CloudWatch Logs から最新のデータを出力して、スケジュールされた呼び出しの結果を表示しています。
+ デモ中に作成されたすべてのリソースをクリーンアップします。
この例は GitHub で最もよく確認できます。完全なソースコードとセットアップおよび実行の手順については、[GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/lambda#readme) で完全な例を参照してください。
**この例で使用されているサービス**
+ CloudWatch Logs
+ DynamoDB
+ EventBridge
+ Lambda
+ Amazon SNS
## サーバーレスサンプル
### Lambda 関数での Amazon RDS データベースへの接続
次のコード例は、RDS データベースに接続する Lambda 関数を実装する方法を示しています。この関数は、シンプルなデータベースリクエストを実行し、結果を返します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用した Lambda 関数での Amazon RDS データベースへの接続
```
import json
import os
import boto3
import pymysql
# RDS settings
proxy_host_name = os.environ['PROXY_HOST_NAME']
port = int(os.environ['PORT'])
db_name = os.environ['DB_NAME']
db_user_name = os.environ['DB_USER_NAME']
aws_region = os.environ['AWS_REGION']
# Fetch RDS Auth Token
def get_auth_token():
client = boto3.client('rds')
token = client.generate_db_auth_token(
DBHostname=proxy_host_name,
Port=port
DBUsername=db_user_name
Region=aws_region
)
return token
def lambda_handler(event, context):
token = get_auth_token()
try:
connection = pymysql.connect(
host=proxy_host_name,
user=db_user_name,
password=token,
db=db_name,
port=port,
ssl={'ca': 'Amazon RDS'} # Ensure you have the CA bundle for SSL connection
)
with connection.cursor() as cursor:
cursor.execute('SELECT %s + %s AS sum', (3, 2))
result = cursor.fetchone()
return result
except Exception as e:
return (f"Error: {str(e)}") # Return an error message if an exception occurs
```
### Kinesis トリガーから Lambda 関数を呼び出す
次のコード例では、Kinesis ストリームからレコードを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。この関数は Kinesis ペイロードを取得し、それを Base64 からデコードして、そのレコードの内容をログ記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用した Lambda での Kinesis イベントの消費。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
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.")
```
### DynamoDB トリガーから Lambda 関数を呼び出す
次のコード例は、DynamoDB ストリームからレコードを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。関数は DynamoDB ペイロードを取得し、レコードの内容をログ記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用して Lambda で DynamoDB イベントの消費。
```
import json
def lambda_handler(event, context):
print(json.dumps(event, indent=2))
for record in event['Records']:
log_dynamodb_record(record)
def log_dynamodb_record(record):
print(record['eventID'])
print(record['eventName'])
print(f"DynamoDB Record: {json.dumps(record['dynamodb'])}")
```
### Amazon DocumentDB トリガーから Lambda 関数を呼び出す
次のコードの例は、DocumentDB 変更ストリームからレコードを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。関数は DocumentDB ペイロードを取得し、レコードの内容をログ記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用して Lambda で Amazon DocumentDB イベントの消費。
```
import json
def lambda_handler(event, context):
for record in event.get('events', []):
log_document_db_event(record)
return 'OK'
def log_document_db_event(record):
event_data = record.get('event', {})
operation_type = event_data.get('operationType', 'Unknown')
db = event_data.get('ns', {}).get('db', 'Unknown')
collection = event_data.get('ns', {}).get('coll', 'Unknown')
full_document = event_data.get('fullDocument', {})
print(f"Operation type: {operation_type}")
print(f"db: {db}")
print(f"collection: {collection}")
print("Full document:", json.dumps(full_document, indent=2))
```
### Amazon MSK トリガーから Lambda 関数を呼び出す
次のコード例は、Amazon MSK クラスターからレコードを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。関数は MSK ペイロードを取得し、レコードの内容をログ記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用した Lambda での Amazon MSK イベントの消費。
```
import base64
def lambda_handler(event, context):
# Iterate through keys
for key in event['records']:
print('Key:', key)
# Iterate through records
for record in event['records'][key]:
print('Record:', record)
# Decode base64
msg = base64.b64decode(record['value']).decode('utf-8')
print('Message:', msg)
```
### Amazon S3 トリガーから Lambda 関数を呼び出す
次のコード例は、S3 バケットにオブジェクトをアップロードすることによってトリガーされるイベントを受け取る Lambda 関数を実装する方法を示しています。この関数は、イベントパラメータから S3 バケット名とオブジェクトキーを取得し、Amazon S3 API を呼び出してオブジェクトのコンテンツタイプを取得してログに記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)リポジトリで完全な例を検索し、設定および実行の方法を確認してください。
Python を使用して Lambda で S3 イベントを消費します。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
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
```
### Amazon SNS トリガーから Lambda 関数を呼び出す
次のコード例は、SNS トピックからメッセージを受信することによってトリガーされるイベントを受け取る Lambda 関数を実装する方法を示しています。この関数はイベントパラメータからメッセージを取得し、各メッセージの内容を記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用して Lambda で SNS イベントを消費します。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
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
```
### Amazon SQS トリガーから Lambda 関数を呼び出す
次のコード例では、SQS キューからメッセージを受信することによってトリガーされるイベントを受け取る、Lambda 関数の実装方法を示しています。この関数はイベントパラメータからメッセージを取得し、各メッセージの内容を記録します。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用した Lambda での SQS イベントの消費。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
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 関数でのバッチアイテム失敗のレポート
以下のコード例では、Kinesis ストリームからイベントを受け取る Lambda 関数のための、部分的なバッチレスポンスの実装方法を示しています。この関数は、レスポンスとしてバッチアイテムの失敗を報告し、対象のメッセージを後で再試行するよう Lambda に伝えます。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用した Lambda での Kinesis バッチアイテム失敗のレポート。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
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":[]}
```
### DynamoDB トリガーで Lambda 関数のバッチアイテムの失敗をレポートする
次のコード例は、DynamoDB ストリームからイベントを受け取る Lambda 関数の部分的なバッチレスポンスの実装方法を示しています。この関数は、レスポンスとしてバッチアイテムの失敗を報告し、対象のメッセージを後で再試行するよう Lambda に伝えます。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用して Lambda で DynamoDB のバッチアイテム失敗のレポート。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def handler(event, context):
records = event.get("Records")
curRecordSequenceNumber = ""
for record in records:
try:
# Process your record
curRecordSequenceNumber = record["dynamodb"]["SequenceNumber"]
except Exception as e:
# Return failed record's sequence number
return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]}
return {"batchItemFailures":[]}
```
### Amazon SQS トリガーを使用した Lambda 関数でのバッチアイテム失敗のレポート
以下のコード例では、SQS キューからイベントを受け取る Lambda 関数のための、部分的なバッチレスポンスの実装方法を示しています。この関数は、レスポンスとしてバッチアイテムの失敗を報告し、対象のメッセージを後で再試行するよう Lambda に伝えます。
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用した Lambda での SQS バッチアイテム失敗のレポート。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
if event:
batch_item_failures = []
sqs_batch_response = {}
for record in event["Records"]:
try:
print(f"Processed message: {record['body']}")
except Exception as e:
batch_item_failures.append({"itemIdentifier": record['messageId']})
sqs_batch_response["batchItemFailures"] = batch_item_failures
return sqs_batch_response
```