def exists(self, name):
name = self._normalize_name(name)
try:
self.bucket.stat_object(name)
except Error, e:
if e[0] == 404:
return False
raise
return True
#def listdir(self, path):
# path = self._normalize_name(path)
# try:
# result = self.bucket.list(path=path)
# return [i.name for i in result]
# except Error, e:
# raise IOError('Storage Error: %s' % e.args)
python类Storage()的实例源码
def _open_read(self, name):
name = self._normalize_name(name)
class _:
def __init__(self, chunks):
self.buf = ''
def read(self, num_bytes=None):
if num_bytes is None:
num_bytes = sys.maxint
try:
while len(self.buf) < num_bytes:
self.buf += chunks.next()
except StopIteration:
pass
except Error, e:
raise IOError('Storage Error: %s' % e.args)
retval = self.buf[:num_bytes]
self.buf = self.buf[num_bytes:]
return retval
chunks = self.bucket.get_object_contents(name, chunk_size=8192)
return _(chunks)
def test_logo_path(self, file_exists, delete_called):
"""
Test that the path of image file should beenterprise/branding/<model.id>/<model_id>_logo.<ext>.lower().
Additionally, test that the correct backend actions are taken in regards to deleting existing data.
"""
file_mock = self._make_file_mock()
branding_config = EnterpriseCustomerBrandingConfiguration(
id=1,
enterprise_customer=factories.EnterpriseCustomerFactory(),
logo=file_mock
)
storage_mock = mock.MagicMock(spec=Storage, name="StorageMock")
storage_mock.exists.return_value = file_exists
with mock.patch("django.core.files.storage.default_storage._wrapped", storage_mock):
path = logo_path(branding_config, branding_config.logo.name)
self.assertEqual(path, "enterprise/branding/1/1_logo.png")
assert storage_mock.delete.call_count == (1 if delete_called else 0)
if delete_called:
storage_mock.delete.assert_called_once_with('enterprise/branding/1/1_logo.png')
def test_branding_configuration_saving_successfully(self):
"""
Test enterprise customer branding configuration saving successfully.
"""
storage_mock = mock.MagicMock(spec=Storage, name="StorageMock")
branding_config_1 = EnterpriseCustomerBrandingConfiguration(
enterprise_customer=factories.EnterpriseCustomerFactory(),
logo="test1.png"
)
storage_mock.exists.return_value = True
with mock.patch("django.core.files.storage.default_storage._wrapped", storage_mock):
branding_config_1.save()
self.assertEqual(EnterpriseCustomerBrandingConfiguration.objects.count(), 1)
branding_config_2 = EnterpriseCustomerBrandingConfiguration(
enterprise_customer=factories.EnterpriseCustomerFactory(),
logo="test2.png"
)
storage_mock.exists.return_value = False
with mock.patch("django.core.files.storage.default_storage._wrapped", storage_mock):
branding_config_2.save()
self.assertEqual(EnterpriseCustomerBrandingConfiguration.objects.count(), 2)
def _get_available_name(self, name, max_length=None):
dir_name, file_name = os.path.split(name)
file_root, file_ext = os.path.splitext(file_name)
while self.exists(name) or (max_length and len(name) > max_length):
# file_ext includes the dot.
name = os.path.join(dir_name, "%s_%s%s" % (file_root, get_random_string(7), file_ext))
if max_length is None:
continue
# Truncate file_root if max_length exceeded.
truncation = len(name) - max_length
if truncation > 0:
file_root = file_root[:-truncation]
# Entire file_root was truncated in attempt to find an available filename.
if not file_root:
raise SuspiciousFileOperation(
'Storage can not find an available filename for "%s". '
'Please make sure that the corresponding file field '
'allows sufficient "max_length".' % name
)
name = os.path.join(dir_name, "%s_%s%s" % (file_root, get_random_string(7), file_ext))
return name
def path(self, name):
"""
Return a local filesystem path where the file can be retrieved using
Python's built-in open() function. Storage systems that can't be
accessed using open() should *not* implement this method.
"""
raise NotImplementedError("This backend doesn't support absolute paths.")
def listdir(self, path):
"""
List the contents of the specified path. Return a 2-tuple of lists:
the first item being directories, the second item being files.
"""
raise NotImplementedError('subclasses of Storage must provide a listdir() method')
def test_storage_check(self):
sm = MagicMock(spec=Storage, name='StorageMock')
with patch('django.core.files.storage.default_storage._wrapped', sm):
check = StorageCheck(name='TestStorageCheck')
check.run_check()
self.assertIn('Data mismatch', check.errors[0])
self.assertEqual(len(check.errors), 1)
def path(self, name):
"""
Returns a local filesystem path where the file can be retrieved using
Python's built-in open() function. Storage systems that can't be
accessed using open() should *not* implement this method.
"""
raise NotImplementedError("This backend doesn't have paths.")
def get_available_name(self, name, max_length=None):
if max_length and len(name) + 41 > max_length:
raise SuspiciousFileOperation(
'Storage can not find an available filename for "%s". '
'Please make sure that the corresponding file field '
'allows sufficient "max_length".' % name
)
return name
def _save(self, name, content):
name = self._normalize_name(name)
try:
self.bucket.put_object(name, content)
except Error, e:
raise IOError('Storage Error: %s' % e.args)
return name
def delete(self, name):
name = self._normalize_name(name)
try:
self.bucket.delete_object(name)
except Error, e:
raise IOError('Storage Error: %s' % e.args)
def size(self, name):
name = self._normalize_name(name)
try:
attrs = self.bucket.stat_object(name)
return attrs.bytes
except Error, e:
raise IOError('Storage Error: %s' % e.args)
def __init__(self, storage=None, *args, **kwargs):
if storage is not None:
self.storage = storage
if self.storage is None:
raise ImproperlyConfigured("The staticfiles storage finder %r "
"doesn't have a storage class "
"assigned." % self.__class__)
# Make sure we have an storage instance here.
if not isinstance(self.storage, (Storage, LazyObject)):
self.storage = self.storage()
super(BaseStorageFinder, self).__init__(*args, **kwargs)
