"""Exceptions module for lambdas. Paired with the error constants.""" from .constants import errors class LoggedException(Exception): """Exception that causes the state transitions in the step function.""" class LambdaError(LoggedException): """Exception raised by lambda during execution with an error code and error params.""" def __init__(self, error_code, error_params): """Construct a LambdaError with an error_code and error_params. Args: error_code (str): Error code used to look up messages in the errors constants. error_params (dict): Dict of values used in constructing the error message. """ self.error_code = error_code error_message = errors.ERROR_MESSAGES.get(error_code, 'Unknown Error').format(**error_params) self.errors = dict() self.errors[error_code] = error_message Exception.__init__(self, error_message) class CloudwatchEventError(LambdaError): """Exception raises if cloudwatch event doesn't have needed arguments.""" def __init__(self, cloudwatch_event): """Construct a CloudwatchEventError with the cloudwatch event. Args: cloudwatch_event (dict): Cloudwatch event """ LambdaError.__init__( self, error_code=errors.CLOUDWATCH_EVENT_ERROR, error_params={'event': cloudwatch_event} ) class PostAssetNotFoundError(LambdaError): """Error for 404 from service and doesnt send an SNS message. Used by acknowledge,general status, final status.""" def __init__(self, response): """Construct a PostAssetNotFoundError with the response object. Args: response (requests.models.Response): Response from the service. """ error_params = { 'code': response.status_code, 'response': response.text } LambdaStatusError.__init__( self, error_code=errors.POST_ASSET_ERROR_CODE, error_params=error_params ) class LambdaStatusError(LambdaError): """Lambda Errors that get logged as asset statuses.""" class S3Error(LambdaStatusError): """Any S3 error: 404, 403 or 500.""" class AssetConfigError(LambdaStatusError): """Error raised when the received configuration from the service is invalid.""" def __init__(self, msg): """Construct an AssetConfigError with the error message. Args: msg (str): Validation error message. """ LambdaStatusError.__init__( self, error_code=errors.ASSET_CONFIG_ERROR, error_params={'message': msg} ) class PostAssetError(PostAssetNotFoundError, LambdaStatusError): """Error when posting to the microserivce. used by acknowledge, general status and final status.""" class LambdaEventError(LambdaStatusError): """Error raised for an invalid lambda event. Use by all steps after acknowledge.""" def __init__(self, msg): """Construct a LambdaEventError with the error message. Args: msg (str): Validation error message. """ LambdaStatusError.__init__( self, error_code=errors.LAMBDA_EVENT_ERROR, error_params={'message': msg} )