我们从Python开源项目中，提取了以下10个代码示例，用于说明如何使用botocore.exceptions.EndpointConnectionError()。
def test_reraise_endpointconnectionerrors_decorator(): @reraise_endpointconnectionerrors def foo(name, age=100): raise EndpointConnectionError(endpoint_url='http://example.com') with pytest.raises(OwnEndpointConnectionError): foo('peter')
def test_check_s3_urls_endpointconnectionerror(botomock, settings): def mock_api_call(self, operation_name, api_params): assert operation_name == 'HeadBucket' if api_params['Bucket'] == 'private': raise EndpointConnectionError(endpoint_url='http://s3.example.com') return {} with botomock(mock_api_call): errors = dockerflow_extra.check_s3_urls(None) assert errors error, = errors assert 'private' in error.msg assert 'EndpointConnectionError' in error.msg
def check_s3_urls(app_configs, **kwargs): errors = [] checked = [] def check_url(url, setting_key): if url in checked: return bucket = S3Bucket(url) if not bucket.private: return try: bucket.s3_client.head_bucket(Bucket=bucket.name) except ClientError as exception: if exception.response['Error']['Code'] in ('403', '404'): errors.append(checks.Error( f'Unable to connect to {url} (bucket={bucket.name!r}, ' f'found in settings.{setting_key}) due to ' f'ClientError ({exception.response!r})', id='tecken.health.E002' )) else: raise except EndpointConnectionError: errors.append(checks.Error( f'Unable to connect to {url} (bucket={bucket.name!r}, ' f'found in settings.{setting_key}) due to ' f'EndpointConnectionError', id='tecken.health.E001' )) else: checked.append(url) for url in settings.SYMBOL_URLS: check_url(url, 'SYMBOL_URLS') for url in settings.UPLOAD_URL_EXCEPTIONS.values(): check_url(url, 'UPLOAD_URL_EXCEPTIONS') return errors
def s3errors(path): """Translate S3 errors to FSErrors.""" try: yield except ClientError as error: _error = error.response.get('Error', {}) error_code = _error.get('Code', None) response_meta = error.response.get('ResponseMetadata', {}) http_status = response_meta.get('HTTPStatusCode', 200) error_msg = _error.get('Message', None) if error_code == 'NoSuchBucket': raise errors.ResourceError(path, exc=error, msg=error_msg) if http_status == 404: raise errors.ResourceNotFound(path) elif http_status == 403: raise errors.PermissionDenied(path=path, msg=error_msg) else: raise errors.OperationFailed(path=path, exc=error) except SSLError as error: raise errors.OperationFailed(path, exc=error) except EndpointConnectionError as error: raise errors.RemoteConnectionError( path, exc=error, msg="{}".format(error) )
def _get_response(self, request, operation_model, attempts): # This will return a tuple of (success_response, exception) # and success_response is itself a tuple of # (http_response, parsed_dict). # If an exception occurs then the success_response is None. # If no exception occurs then exception is None. try: logger.debug("Sending http request: %s", request) http_response = self.http_session.send( request, verify=self.verify, stream=operation_model.has_streaming_output, proxies=self.proxies, timeout=self.timeout) except ConnectionError as e: # For a connection error, if it looks like it's a DNS # lookup issue, 99% of the time this is due to a misconfigured # region/endpoint so we'll raise a more specific error message # to help users. logger.debug("ConnectionError received when sending HTTP request.", exc_info=True) if self._looks_like_dns_error(e): endpoint_url = e.request.url better_exception = EndpointConnectionError( endpoint_url=endpoint_url, error=e) return (None, better_exception) elif self._looks_like_bad_status_line(e): better_exception = ConnectionClosedError( endpoint_url=e.request.url, request=e.request) return (None, better_exception) else: return (None, e) except Exception as e: logger.debug("Exception received when sending HTTP request.", exc_info=True) return (None, e) # This returns the http_response and the parsed_data. response_dict = convert_to_response_dict(http_response, operation_model) parser = self._response_parser_factory.create_parser( operation_model.metadata['protocol']) parsed_response = parser.parse( response_dict, operation_model.output_shape) return (http_response, parsed_response), None
