Python code snippet for logging failed AWS API calls with boto3 / botocore

botocore_error_logger.py

import logging

import boto3

def _log_botocore_errors(*_args, **kwargs):
    """Handler for botocore needs-retry events."""
    # Arg parsing as per
    # https://github.com/boto/botocore/blob/6451ae1fad57f4453af97649e7ed9192b0f623be/botocore/retries/standard.py#L106
    attempt_number = kwargs["attempts"]
    operation_model = kwargs["operation"]
    operation_name = operation_model.name

    response = kwargs["response"]
    if response is None:
        parsed_response = None
    else:
        _, parsed_response = response

    caught_exception = kwargs["caught_exception"]
    if parsed_response is None and caught_exception:
        logging.warning(
            "%s failed with exception (attempt %i)",
            operation_name,
            attempt_number,
            exc_info=caught_exception,
        )
        return

    if parsed_response is None:
        # No response or exception?
        return

    error = parsed_response.get("Error", {})
    if not error:
        # All good
        return

    metadata = parsed_response.get("ResponseMetadata", {})
    status_code = metadata.get("HTTPStatusCode")
    code = error.get("Code")
    message = error.get("Message")
    logging.warning(
        "%s error: %s - %s (status %s; attempt %i)",
        operation_name,
        code,
        message,
        status_code,
        attempt_number,
    )


def _register_error_logger_for_resource(resource):
    """Register error logger for botocore service resource.

    Args:
        resource: Service resource.
    """
    _register_error_logger_for_client(resource.meta.client)

def _register_error_logger_for_client(client):
    """Register error logger for botocore client.

    Args:
        client: Client.
    """
    client.meta.events.register(
        "needs-retry", _log_botocore_errors
    )

# Example usage
ddb_resource = boto3.resource("dynamodb")
ddb_client = boto3.client("dynamodb")

logging.basicConfig(level=logging.WARNING)

_register_error_logger_for_resource(ddb_resource)
_register_error_logger_for_client(ddb_client)

ddb_client.list_tables()