2020-07-09 17:27:30 -07:00
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Purpose
2022-04-08 15:42:13 -07:00
Shows how to use the AWS SDK for Python (Boto3) to manage and invoke AWS Lambda
functions.
2020-07-09 17:27:30 -07:00
"""
import io
import json
import logging
import zipfile
from botocore . exceptions import ClientError
logger = logging . getLogger ( __name__ )
2022-04-08 15:42:13 -07:00
# 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 ) :
"""
2022-04-12 12:01:44 -07:00
Creates a Lambda deployment package in .zip format in an in-memory buffer. This
2022-04-08 15:42:13 -07:00
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
2020-07-09 17:27:30 -07:00
try :
2022-04-08 15:42:13 -07:00
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 )
2020-07-09 17:27:30 -07:00
except ClientError as error :
2022-04-08 15:42:13 -07:00
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 )
2020-07-09 17:27:30 -07:00
else :
2022-04-08 15:42:13 -07:00
logger . exception (
" Couldn ' t create role %s or attach policy %s . " ,
iam_role_name , policy_arn )
2020-07-09 17:27:30 -07:00
raise
2022-04-08 15:42:13 -07:00
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
2022-04-12 12:01:44 -07:00
code in .zip format.
2022-04-08 15:42:13 -07:00
: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 )
2020-07-09 17:27:30 -07:00
raise
2022-04-08 15:42:13 -07:00
else :
return function_arn
# snippet-end:[python.example_code.lambda.CreateFunction]
2020-07-09 17:27:30 -07:00
2022-04-08 15:42:13 -07:00
# 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.
2022-04-12 12:01:44 -07:00
:param get_log: When true, the last 4 KB of the execution log are included in
2022-04-08 15:42:13 -07:00
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 ) :
"""
2022-04-12 12:01:44 -07:00
Updates the code for a Lambda function by submitting a .zip archive that contains
2022-04-08 15:42:13 -07:00
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
2022-04-12 12:01:44 -07:00
.zip format.
2022-04-08 15:42:13 -07:00
: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]
