# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 """ Purpose Shows how to use the AWS SDK for Python (Boto3) to manage and invoke AWS Lambda functions. """ import io import json import logging import zipfile from botocore.exceptions import ClientError logger = logging.getLogger(__name__) # snippet-start:[python.example_code.python.LambdaWrapper.full] # snippet-start:[python.example_code.python.LambdaWrapper.decl] class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource # snippet-end:[python.example_code.python.LambdaWrapper.decl] @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 # snippet-start:[python.example_code.lambda.GetFunction] 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 # snippet-end:[python.example_code.lambda.GetFunction] # snippet-start:[python.example_code.lambda.CreateFunction] 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 # snippet-end:[python.example_code.lambda.CreateFunction] # snippet-start:[python.example_code.lambda.DeleteFunction] 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 # snippet-end:[python.example_code.lambda.DeleteFunction] # snippet-start:[python.example_code.lambda.Invoke] 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 # snippet-end:[python.example_code.lambda.Invoke] # snippet-start:[python.example_code.lambda.UpdateFunctionCode] 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 # snippet-end:[python.example_code.lambda.UpdateFunctionCode] # snippet-start:[python.example_code.lambda.UpdateFunctionConfiguration] 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 # snippet-end:[python.example_code.lambda.UpdateFunctionConfiguration] # snippet-start:[python.example_code.lambda.ListFunctions] 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 # snippet-end:[python.example_code.lambda.ListFunctions] # snippet-end:[python.example_code.python.LambdaWrapper.full]