codekingpro/portable-devtools

1"""Special cased retries.2 3These are additional retry cases we still have to handle from the legacy4retry handler.  They don't make sense as part of the standard mode retry5module.  Ideally we should be able to remove this module.6 7"""8import logging9from binascii import crc3210 11from botocore.retries.base import BaseRetryableChecker12 13logger = logging.getLogger(__name__)14 15 16# TODO: This is an ideal candidate for the retryable trait once that's17# available.18class RetryIDPCommunicationError(BaseRetryableChecker):19    _SERVICE_NAME = 'sts'20 21    def is_retryable(self, context):22        service_name = context.operation_model.service_model.service_name23        if service_name != self._SERVICE_NAME:24            return False25        error_code = context.get_error_code()26        return error_code == 'IDPCommunicationError'27 28 29class RetryDDBChecksumError(BaseRetryableChecker):30    _CHECKSUM_HEADER = 'x-amz-crc32'31    _SERVICE_NAME = 'dynamodb'32 33    def is_retryable(self, context):34        service_name = context.operation_model.service_model.service_name35        if service_name != self._SERVICE_NAME:36            return False37        if context.http_response is None:38            return False39        checksum = context.http_response.headers.get(self._CHECKSUM_HEADER)40        if checksum is None:41            return False42        actual_crc32 = crc32(context.http_response.content) & 0xFFFFFFFF43        if actual_crc32 != int(checksum):44            logger.debug(45                "DynamoDB crc32 checksum does not match, "46                "expected: %s, actual: %s",47                checksum,48                actual_crc32,49            )50            return True51