Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class MaxFileSizeExceeded(B2Error):
def __init__(self, size, max_allowed_size):
super(MaxFileSizeExceeded, self).__init__()
self.size = size
self.max_allowed_size = max_allowed_size
def __str__(self):
return 'Allowed file size of exceeded: %s > %s' % (
self.size,
self.max_allowed_size,
)
class MaxRetriesExceeded(B2Error):
def __init__(self, limit, exception_info_list):
super(MaxRetriesExceeded, self).__init__()
self.limit = limit
self.exception_info_list = exception_info_list
def __str__(self):
exceptions = '\n'.join(str(wrapped_error) for wrapped_error in self.exception_info_list)
return 'FAILED to upload after %s tries. Encountered exceptions: %s' % (
self.limit,
exceptions,
)
class MissingPart(B2SimpleError):
prefix = 'Part number has not been uploaded'
def __init__(self, message, code):
super(InvalidAuthToken,
self).__init__('Invalid authorization token. Server said: ' + message, code)
class RestrictedBucket(B2Error):
def __init__(self, bucket_name):
super(RestrictedBucket, self).__init__()
self.bucket_name = bucket_name
def __str__(self):
return 'Application key is restricted to bucket: %s' % self.bucket_name
class MaxFileSizeExceeded(B2Error):
def __init__(self, size, max_allowed_size):
super(MaxFileSizeExceeded, self).__init__()
self.size = size
self.max_allowed_size = max_allowed_size
def __str__(self):
return 'Allowed file size of exceeded: %s > %s' % (
self.size,
self.max_allowed_size,
)
class MaxRetriesExceeded(B2Error):
def __init__(self, limit, exception_info_list):
super(MaxRetriesExceeded, self).__init__()
self.limit = limit
class UnexpectedCloudBehaviour(B2SimpleError):
pass
class UnknownError(B2SimpleError):
pass
class UnknownHost(B2Error):
def __str__(self):
return 'unknown host'
class UnrecognizedBucketType(B2Error):
pass
class UnsatisfiableRange(B2Error):
def __str__(self):
return "The range in the request is outside the size of the file"
class UploadTokenUsedConcurrently(B2Error):
def __init__(self, token):
super(UploadTokenUsedConcurrently, self).__init__()
self.token = token
def __str__(self):
return "More than one concurrent upload using auth token %s" % (self.token,)
def __str__(self):
return 'only %d of %d bytes read' % (
self.bytes_read,
self.file_size,
)
class UnexpectedCloudBehaviour(B2SimpleError):
pass
class UnknownError(B2SimpleError):
pass
class UnknownHost(B2Error):
def __str__(self):
return 'unknown host'
class UnrecognizedBucketType(B2Error):
pass
class UnsatisfiableRange(B2Error):
def __str__(self):
return "The range in the request is outside the size of the file"
class UploadTokenUsedConcurrently(B2Error):
def __init__(self, token):
super(UploadTokenUsedConcurrently, self).__init__()
class UnusableFileName(B2SimpleError):
"""
Raise when a filename doesn't meet the rules.
Could possibly use InvalidUploadSource, but this is intended for the filename on the
server, which could differ. https://www.backblaze.com/b2/docs/files.html.
"""
pass
class InvalidMetadataDirective(B2Error):
pass
class InvalidRange(B2Error):
def __init__(self, content_length, range_):
super(InvalidRange, self).__init__()
self.content_length = content_length
self.range_ = range_
def __str__(self):
return 'A range of %d-%d was requested (size of %d), but cloud could only serve %d of that' % (
self.range_[0],
self.range_[1],
self.range_[1] - self.range_[0] + 1,
self.content_length,
)
class InvalidUploadSource(B2SimpleError):
pass
"""
Return true if this is an error that can cause an HTTP
call to be retried.
"""
return False
def should_retry_upload(self):
"""
Return true if this is an error that should tell the upload
code to get a new upload URL and try the upload again.
"""
return False
@six.add_metaclass(ABCMeta)
class B2SimpleError(B2Error):
"""
A B2Error with a message prefix.
"""
def __str__(self):
return '%s: %s' % (self.prefix, super(B2SimpleError, self).__str__())
@six.add_metaclass(ABCMeta)
class NotAllowedByAppKeyError(B2SimpleError):
"""
Base class for errors caused by restrictions on an application key.
"""
@six.add_metaclass(ABCMeta)
return 'Cannot upload files, storage cap exceeded.'
class TooManyRequests(B2Error):
def __init__(self, retry_after_seconds=None):
super(TooManyRequests, self).__init__()
self.retry_after_seconds = retry_after_seconds
def __str__(self):
return 'Too many requests'
def should_retry_http(self):
return True
class TruncatedOutput(TransientErrorMixin, B2Error):
def __init__(self, bytes_read, file_size):
super(TruncatedOutput, self).__init__()
self.bytes_read = bytes_read
self.file_size = file_size
def __str__(self):
return 'only %d of %d bytes read' % (
self.bytes_read,
self.file_size,
)
class UnexpectedCloudBehaviour(B2SimpleError):
pass
######################################################################
#
# File: b2sdk/sync/exception.py
#
# Copyright 2019 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################
from ..exception import B2Error, B2SimpleError
class EnvironmentEncodingError(B2Error):
"""
Raised when a file name can not be decoded with system encoding.
"""
def __init__(self, filename, encoding):
"""
:param filename: an encoded file name
:type filename: str, bytes
:param str encoding: file name encoding
"""
super(EnvironmentEncodingError, self).__init__()
self.filename = filename
self.encoding = encoding
def __str__(self):
return """file name %s cannot be decoded with system encoding (%s).
class UnknownHost(B2Error):
def __str__(self):
return 'unknown host'
class UnrecognizedBucketType(B2Error):
pass
class UnsatisfiableRange(B2Error):
def __str__(self):
return "The range in the request is outside the size of the file"
class UploadTokenUsedConcurrently(B2Error):
def __init__(self, token):
super(UploadTokenUsedConcurrently, self).__init__()
self.token = token
def __str__(self):
return "More than one concurrent upload using auth token %s" % (self.token,)
def interpret_b2_error(status, code, message, response_headers, post_params=None):
post_params = post_params or {}
if status == 400 and code == "already_hidden":
return FileAlreadyHidden(post_params.get('fileName'))
elif status == 400 and code == 'bad_json':
return BadJson(message)
elif (
(status == 400 and code in ("no_such_file", "file_not_present")) or
class PartSha1Mismatch(B2Error):
def __init__(self, key):
super(PartSha1Mismatch, self).__init__()
self.key = key
def __str__(self):
return 'Part number %s has wrong SHA1' % (self.key,)
class ServiceError(TransientErrorMixin, B2Error):
"""
Used for HTTP status codes 500 through 599.
"""
class StorageCapExceeded(B2Error):
def __str__(self):
return 'Cannot upload files, storage cap exceeded.'
class TooManyRequests(B2Error):
def __init__(self, retry_after_seconds=None):
super(TooManyRequests, self).__init__()
self.retry_after_seconds = retry_after_seconds
def __str__(self):
return 'Too many requests'
def should_retry_http(self):
return True