Python botocore.exceptions 模块，ChecksumError() 实例源码

我们从Python开源项目中，提取了以下12个代码示例，用于说明如何使用botocore.exceptions.ChecksumError()。

项目：aws-cfn-plex 作者：lordmuffin | 项目源码 | 文件源码

def _extract_retryable_exception(config):
    applies_when = config['applies_when']
    if 'crc32body' in applies_when.get('response', {}):
        return [ChecksumError]
    elif 'socket_errors' in applies_when:
        exceptions = []
        for name in applies_when['socket_errors']:
            exceptions.extend(EXCEPTION_MAP[name])
        return exceptions

项目：aws-cfn-plex 作者：lordmuffin | 项目源码 | 文件源码

def _check_response(self, attempt_number, response):
        http_response = response[0]
        expected_crc = http_response.headers.get(self._header_name)
        if expected_crc is None:
            logger.debug("crc32 check skipped, the %s header is not "
                         "in the http response.", self._header_name)
        else:
            actual_crc32 = crc32(response[0].content) & 0xffffffff
            if not actual_crc32 == int(expected_crc):
                logger.debug(
                    "retry needed: crc32 check failed, expected != actual: "
                    "%s != %s", int(expected_crc), actual_crc32)
                raise ChecksumError(checksum_type='crc32',
                                    expected_checksum=int(expected_crc),
                                    actual_checksum=actual_crc32)

项目：AshsSDK 作者：thehappydinoa | 项目源码 | 文件源码

def _extract_retryable_exception(config):
    applies_when = config['applies_when']
    if 'crc32body' in applies_when.get('response', {}):
        return [ChecksumError]
    elif 'socket_errors' in applies_when:
        exceptions = []
        for name in applies_when['socket_errors']:
            exceptions.extend(EXCEPTION_MAP[name])
        return exceptions

项目：AshsSDK 作者：thehappydinoa | 项目源码 | 文件源码

def _check_response(self, attempt_number, response):
        http_response = response[0]
        expected_crc = http_response.headers.get(self._header_name)
        if expected_crc is None:
            logger.debug("crc32 check skipped, the %s header is not "
                         "in the http response.", self._header_name)
        else:
            actual_crc32 = crc32(response[0].content) & 0xffffffff
            if not actual_crc32 == int(expected_crc):
                logger.debug(
                    "retry needed: crc32 check failed, expected != actual: "
                    "%s != %s", int(expected_crc), actual_crc32)
                raise ChecksumError(checksum_type='crc32',
                                    expected_checksum=int(expected_crc),
                                    actual_checksum=actual_crc32)

项目：aws-ec2rescue-linux 作者：awslabs | 项目源码 | 文件源码

def _extract_retryable_exception(config):
    applies_when = config['applies_when']
    if 'crc32body' in applies_when.get('response', {}):
        return [ChecksumError]
    elif 'socket_errors' in applies_when:
        exceptions = []
        for name in applies_when['socket_errors']:
            exceptions.extend(EXCEPTION_MAP[name])
        return exceptions

项目：aws-ec2rescue-linux 作者：awslabs | 项目源码 | 文件源码

def _check_response(self, attempt_number, response):
        http_response = response[0]
        expected_crc = http_response.headers.get(self._header_name)
        if expected_crc is None:
            logger.debug("crc32 check skipped, the %s header is not "
                         "in the http response.", self._header_name)
        else:
            actual_crc32 = crc32(response[0].content) & 0xffffffff
            if not actual_crc32 == int(expected_crc):
                logger.debug(
                    "retry needed: crc32 check failed, expected != actual: "
                    "%s != %s", int(expected_crc), actual_crc32)
                raise ChecksumError(checksum_type='crc32',
                                    expected_checksum=int(expected_crc),
                                    actual_checksum=actual_crc32)

项目：jepsen-training-vpc 作者：bloomberg | 项目源码 | 文件源码

def _extract_retryable_exception(config):
    applies_when = config['applies_when']
    if 'crc32body' in applies_when.get('response', {}):
        return [ChecksumError]
    elif 'socket_errors' in applies_when:
        exceptions = []
        for name in applies_when['socket_errors']:
            exceptions.extend(EXCEPTION_MAP[name])
        return exceptions

项目：jepsen-training-vpc 作者：bloomberg | 项目源码 | 文件源码

def _check_response(self, attempt_number, response):
        http_response = response[0]
        expected_crc = http_response.headers.get(self._header_name)
        if expected_crc is None:
            logger.debug("crc32 check skipped, the %s header is not "
                         "in the http response.", self._header_name)
        else:
            actual_crc32 = crc32(response[0].content) & 0xffffffff
            if not actual_crc32 == int(expected_crc):
                logger.debug(
                    "retry needed: crc32 check failed, expected != actual: "
                    "%s != %s", int(expected_crc), actual_crc32)
                raise ChecksumError(checksum_type='crc32',
                                    expected_checksum=int(expected_crc),
                                    actual_checksum=actual_crc32)

项目：AWS-AutoTag 作者：cpollard0 | 项目源码 | 文件源码

def _extract_retryable_exception(config):
    applies_when = config['applies_when']
    if 'crc32body' in applies_when.get('response', {}):
        return [ChecksumError]
    elif 'socket_errors' in applies_when:
        exceptions = []
        for name in applies_when['socket_errors']:
            exceptions.extend(EXCEPTION_MAP[name])
        return exceptions

项目：AWS-AutoTag 作者：cpollard0 | 项目源码 | 文件源码

def _check_response(self, attempt_number, response):
        http_response = response[0]
        expected_crc = http_response.headers.get(self._header_name)
        if expected_crc is None:
            logger.debug("crc32 check skipped, the %s header is not "
                         "in the http response.", self._header_name)
        else:
            actual_crc32 = crc32(response[0].content) & 0xffffffff
            if not actual_crc32 == int(expected_crc):
                logger.debug(
                    "retry needed: crc32 check failed, expected != actual: "
                    "%s != %s", int(expected_crc), actual_crc32)
                raise ChecksumError(checksum_type='crc32',
                                    expected_checksum=int(expected_crc),
                                    actual_checksum=actual_crc32)

项目：tf_aws_ecs_instance_draining_on_scale_in 作者：terraform-community-modules | 项目源码 | 文件源码

def _extract_retryable_exception(config):
    applies_when = config['applies_when']
    if 'crc32body' in applies_when.get('response', {}):
        return [ChecksumError]
    elif 'socket_errors' in applies_when:
        exceptions = []
        for name in applies_when['socket_errors']:
            exceptions.extend(EXCEPTION_MAP[name])
        return exceptions

项目：tf_aws_ecs_instance_draining_on_scale_in 作者：terraform-community-modules | 项目源码 | 文件源码

def _check_response(self, attempt_number, response):
        http_response = response[0]
        expected_crc = http_response.headers.get(self._header_name)
        if expected_crc is None:
            logger.debug("crc32 check skipped, the %s header is not "
                         "in the http response.", self._header_name)
        else:
            actual_crc32 = crc32(response[0].content) & 0xffffffff
            if not actual_crc32 == int(expected_crc):
                logger.debug(
                    "retry needed: crc32 check failed, expected != actual: "
                    "%s != %s", int(expected_crc), actual_crc32)
                raise ChecksumError(checksum_type='crc32',
                                    expected_checksum=int(expected_crc),
                                    actual_checksum=actual_crc32)