Error matching for botocore

aws_error_utils.py

# This is a now proper library (that can still be copied into your project as a single file)
# https://github.com/benkehoe/aws-error-utils

# MIT No Attribution
#
# Copyright 2022 Ben Kehoe
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this
# software and associated documentation files (the "Software"), to deal in the Software
# without restriction, including without limitation the rights to use, copy, modify,
# merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

"""
Snippets:
error_code = e.response.get('Error', {}).get('Code')
error_msg = e.response.get('Error', {}).get('Message')
error_status_code = e.response.get('ResponseMetadata', {}).get('HTTPStatusCode')
operation = e.operation_name
"""

import collections
AWSErrorInfo = collections.namedtuple('AWSErrorInfo', ['code', 'message', 'http_status_code', 'operation_name', 'response'])
def get_aws_error_info(client_error):
    """Returns an AWSErrorInfo namedtuple with the important details of the error extracted"""
    if not isinstance(client_error, botocore.exceptions.ClientError):
        raise TypeError("Error is of type {}, not ClientError".format(client_error))
    return AWSErrorInfo(
            client_error.response.get('Error', {}).get('Code'),
            client_error.response.get('Error', {}).get('Message'),
            client_error.response.get('ResponseMetadata', {}).get('HTTPStatusCode'),
            client_error.operation_name,
            client_error.response,
    )

def aws_error_matches(client_error, *args, **kwargs):
    """Tests if a botocore.exceptions.ClientError matches the arguments.
    Any positional arguments and the contents of the 'code' kwarg are matched
    against the Error.Code response field.
    If the 'operation_name' kwarg is provided, it is matched against the
    operation_name property.
    Both kwargs can either be a single string or a list of strings.
    You can provide "*" to match any error or any operation.

    try:
        s3 = boto3.client('s3')
        s3.list_objects_v2(Bucket='bucket-1')
        s3.get_object(Bucket='bucket-2', Key='example')
    except botocore.exceptions.ClientError as e:
        if aws_error_matches(e, 'NoSuchBucket', operation_name='GetObject'):
            pass
        else:
            raise
    """
    err_args = args + tuple((kwargs.get('code'),) if isinstance(kwargs.get('code'), str) else kwargs.get('code', tuple()))
    op_args = (kwargs.get('operation_name'),) if isinstance(kwargs.get('operation_name'), str) else kwargs.get('operation_name', tuple())
    if not err_args:
        raise ValueError('No error codes provided')
    err = client_error.response.get('Error', {}).get('Code')
    err_matches = (err and (err in err_args)) or ("*" in err_args)
    op_matches = (client_error.operation_name in op_args) or (not op_args) or ("*" in op_args)
    return err_matches and op_matches

def catch_aws_error(*args, **kwargs):
    """For use in an except statement, returns the current error if it matches the arguments, otherwise a non-matching error
    Any positional arguments and the contents of the 'code' kwarg are matched
    against the Error.Code response field.
    If the 'operation_name' kwarg is provided, it is matched against the
    operation_name property.
    Both kwargs can either be a single string or a list of strings.
    You can provide "*" to match any error or any operation.
    Alternatively, provide a callable that takes the error and returns true for a match.

    try:
        s3 = boto3.client('s3')
        s3.list_objects_v2(Bucket='bucket-1')
        s3.get_object(Bucket='bucket-2', Key='example')
    except catch_aws_error('NoSuchBucket', operation_name='GetObject'):
        pass
    """
    import sys
    import botocore.exceptions
    client_error = sys.exc_info()[1]
    if isinstance(client_error, botocore.exceptions.ClientError):
        if len(args) == 1 and callable(args[0]):
            if args[0](client_error):
                return type(client_error)
        else:
            err_args = args + tuple((kwargs.get('code'),) if isinstance(kwargs.get('code'), str) else kwargs.get('code', tuple()))
            op_args = (kwargs.get('operation_name'),) if isinstance(kwargs.get('operation_name'), str) else kwargs.get('operation_name', tuple())
            if not err_args:
                raise ValueError('No error codes provided')
            err = client_error.response.get('Error', {}).get('Code')
            err_matches = (err and (err in err_args)) or ("*" in err_args)
            op_matches = (client_error.operation_name in op_args) or (not op_args) or ("*" in op_args)
            if err_matches and op_matches:
                return type(client_error)
    return type('RedHerring', (BaseException,), {})