"""Exceptions module for lambdas.""" from src import status from constants import errors class UnexpectedEventBody(Exception): """ Exceptions that raises if input event doesn't have needed arguments. Attributes: expected_args (list): Expected event body arguments. """ def __init__(self, expected_args): """Show exception message. Args: expected_args (list): Expected event body arguments. """ args_sting = ' '.join(expected_args) Exception.__init__( self, ("Expected event arguments: {args} haven\'t been provided" .format(args=args_sting)) ) class LoggedException(Exception): """Exception to avoid propagation of general notification messages.""" class StopProcessingException(Exception): """Exception raised if asset processing must be stopped.""" class AssetUploadAlreadyHandled(Exception): """Exception raised if asset upload has already been handled.""" def notify_and_raise( function, error_status, error_code, filename, bucket, error_params=None, input_params=None): """Send status notification and raise expection. Args: function (str): Lambda function name that causes an error. error_status (str): Error status for lambda function. error_code (str): Error code. filename (str): Current processed asset filename. bucket (str): Current processed asset source bucket name. error_params (dict): Additional error params. input_params (dict): Lambda additional input params for reporting. """ if not error_params: error_params = {} error_message = errors.ERROR_MESSAGES.get( error_code, 'Unknown Error').format(**error_params) status.send_general_status( function=function, status=error_status, filename=filename, error_code=error_code, description=error_message, bucket=bucket, input_params=input_params, ) raise LoggedException(error_message)