Lambda examples using SDK for Python (Boto3) - AWS SDK Code Examples

There are more AWS SDK examples available in the AWS Doc SDK Examples GitHub repo.

Lambda examples using SDK for Python (Boto3)

The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Python (Boto3) with Lambda.

Actions are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios and cross-service examples.

Scenarios are code examples that show you how to accomplish a specific task by calling multiple functions within the same service.

Each example includes a link to GitHub, where you can find instructions on how to set up and run the code in context.

Actions

The following code example shows how to create a Lambda function.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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
  • For API details, see CreateFunction in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to delete a Lambda function.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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
  • For API details, see DeleteFunction in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to get a Lambda function.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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
  • For API details, see GetFunction in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to invoke a Lambda function.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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
  • For API details, see Invoke in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to list Lambda functions.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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
  • For API details, see ListFunctions in AWS SDK for Python (Boto3) API Reference.

The following code example shows how to update Lambda function code.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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

The following code example shows how to update Lambda function configuration.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

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

Scenarios

The following code example shows how to:

  • Create an IAM role and Lambda function, then upload handler code.

  • Invoke the function with a single parameter and get results.

  • Update the function code and configure with an environment variable.

  • Invoke the function with new parameters and get results. Display the returned execution log.

  • List the functions for your account, then clean up resources.

For more information, see Create a Lambda function with the console.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

Define a Lambda handler that increments a number.

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

Define a second Lambda handler that performs arithmetic operations.

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

Create functions that wrap Lambda actions.

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

Create a function that runs the scenario.

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!")

Serverless examples

The following code example shows how to implement a Lambda function that receives an event triggered by receiving records from a Kinesis stream. The function retrieves the Kinesis payload, decodes from Base64, and logs the record contents.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Consuming a Kinesis event with Lambda using Python.

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.")

The following code example shows how to implement a Lambda function that receives an event triggered by uploading an object to an S3 bucket. The function retrieves the S3 bucket name and object key from the event parameter and calls the Amazon S3 API to retrieve and log the content type of the object.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Consuming an S3 event with Lambda using Python.

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

The following code example shows how to implement a Lambda function that receives an event triggered by receiving messages from an SNS topic. The function retrieves the messages from the event parameter and logs the content of each message.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Consuming an SNS event with Lambda using Python.

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

The following code example shows how to implement a Lambda function that receives an event triggered by receiving messages from an SQS queue. The function retrieves the messages from the event parameter and logs the content of each message.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Consuming an SQS event with Lambda using Python.

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

The following code example shows how to implement partial batch response for Lambda functions that receive events from a Kinesis stream. The function reports the batch item failures in the response, signaling to Lambda to retry those messages later.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Reporting Kinesis batch item failures with Lambda using Python.

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":[]}

The following code example shows how to implement partial batch response for Lambda functions that receive events from an SQS queue. The function reports the batch item failures in the response, signaling to Lambda to retry those messages later.

SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the Serverless examples repository.

Reporting SQS batch item failures with Lambda using Python.

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