"""Lambda function SDK utils.""" import json import boto3 from botocore.exceptions import ClientError from lambdacommon.common_config import logger from lambdacommon.common_config import AWS_DEFAULT_REGION LAMBDA_CLIENT = boto3.client('lambda', region_name=AWS_DEFAULT_REGION) class LambdaFunction: """Encapsulates Lambda Function actions.""" function_name: str def __init__(self, name: str): """Init a LambdaFunction class.""" self.function_name = name def invoke_synchronously(self, payload: dict, get_log=None) -> dict: """Invoke Lambda Synchronously. Args: payload: Lambda payload. get_log: Set to include execution log. Returns: Lambda response and info. """ try: response = LAMBDA_CLIENT.invoke( FunctionName=self.function_name, Payload=json.dumps(payload), LogType='Tail' if get_log else 'None', InvocationType='RequestResponse' ) logger.info(f'Invoked function {self.function_name}') except ClientError as err: logger.error( "Couldn't invoke function %s. Reason: %s: %s", self.function_name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response def invoke_asynchronously(self, payload: dict) -> dict: """Invoke Lambda Asynchronously. Args: payload: Lambda payload. Returns: Lambda Status. This doesn't return a response payload. """ try: response = LAMBDA_CLIENT.invoke( FunctionName=self.function_name, Payload=json.dumps(payload), LogType='None', InvocationType='Event' ) logger.info(f'Invoked function {self.function_name}') except ClientError as err: logger.error( "Couldn't invoke function %s. Reason: %s: %s", self.function_name, err.response['Error']['Code'], err.response['Error']['Message']) raise else: return response