import hashlib
import binascii
-from base import DEFAULT_QUOTA, DEFAULT_VERSIONING, NotAllowedError, QuotaError, BaseBackend
+from base import DEFAULT_QUOTA, DEFAULT_VERSIONING, NotAllowedError, QuotaError, BaseBackend, \
+ AccountExists, ContainerExists, AccountNotEmpty, ContainerNotEmpty, ItemNotExists, VersionNotExists
# Stripped-down version of the HashMap class found in tools.
class HashMap(list):
raise NotAllowedError
node = self.node.node_lookup(account)
if node is not None:
- raise NameError('Account already exists')
+ raise AccountExists('Account already exists')
if policy:
self._check_policy(policy)
node = self._put_path(user, self.ROOTNODE, account)
if node is None:
return
if not self.node.node_remove(node):
- raise IndexError('Account is not empty')
+ raise AccountNotEmpty('Account is not empty')
self.permissions.group_destroy(account)
@backend_method
start, limit = self._list_limits(allowed, marker, limit)
return allowed[start:start + limit]
if shared or public:
- allowed = []
+ allowed = set()
if shared:
- allowed.extend([x.split('/', 2)[1] for x in self.permissions.access_list_shared(account)])
+ allowed.update([x.split('/', 2)[1] for x in self.permissions.access_list_shared(account)])
if public:
- allowed.extend([x[0].split('/', 2)[1] for x in self.permissions.public_list(account)])
- allowed = list(set(allowed))
- allowed.sort()
+ allowed.update([x[0].split('/', 2)[1] for x in self.permissions.public_list(account)])
+ allowed = sorted(allowed)
start, limit = self._list_limits(allowed, marker, limit)
return allowed[start:start + limit]
node = self.node.node_lookup(account)
- return [x[0] for x in self._list_object_properties(node, account, '', '/', marker, limit, False, None, [], until)]
+ containers = [x[0] for x in self._list_object_properties(node, account, '', '/', marker, limit, False, None, [], until)]
+ start, limit = self._list_limits([x[0] for x in containers], marker, limit)
+ return containers[start:start + limit]
@backend_method
def list_container_meta(self, user, account, container, domain, until=None):
except NameError:
pass
else:
- raise NameError('Container already exists')
+ raise ContainerExists('Container already exists')
if policy:
self._check_policy(policy)
path = '/'.join((account, container))
for h in hashes:
self.store.map_delete(h)
self.node.node_purge_children(node, until, CLUSTER_DELETED)
- self._report_size_change(user, account, -size, {'action': 'container purge'})
+ self._report_size_change(user, account, -size, {'action': 'container purge', 'path':path})
return
- if self._get_statistics(node)[0] > 0:
- raise IndexError('Container is not empty')
- hashes, size = self.node.node_purge_children(node, inf, CLUSTER_HISTORY)
- for h in hashes:
- self.store.map_delete(h)
- self.node.node_purge_children(node, inf, CLUSTER_DELETED)
- self.node.node_remove(node)
- self._report_size_change(user, account, -size, {'action': 'container delete'})
+ if not delimiter:
+ if self._get_statistics(node)[0] > 0:
+ raise ContainerNotEmpty('Container is not empty')
+ hashes, size = self.node.node_purge_children(node, inf, CLUSTER_HISTORY)
+ for h in hashes:
+ self.store.map_delete(h)
+ self.node.node_purge_children(node, inf, CLUSTER_DELETED)
+ self.node.node_remove(node)
+ self._report_size_change(user, account, -size, {'action': 'container delete', 'path':path})
+ else:
+ # remove only contents
+ src_names = self._list_objects_no_limit(user, account, container, prefix='', delimiter=None, virtual=False, domain=None, keys=[], shared=False, until=None, size_range=None, all_props=True, public=False)
+ paths = []
+ for t in src_names:
+ path = '/'.join((account, container, t[0]))
+ node = t[2]
+ src_version_id, dest_version_id = self._put_version_duplicate(user, node, size=0, type='', hash=None, checksum='', cluster=CLUSTER_DELETED)
+ del_size = self._apply_versioning(account, container, src_version_id)
+ if del_size:
+ self._report_size_change(user, account, -del_size, {'action': 'object delete', 'path':path})
+ self._report_object_change(user, account, path, details={'action': 'object delete'})
+ paths.append(path)
+ self.permissions.access_clear_bulk(paths)
def _list_objects(self, user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, all_props, public):
if user != account and until:
raise NotAllowedError
+ if shared and public:
+ # get shared first
+ shared = self._list_object_permissions(user, account, container, prefix, shared=True, public=False)
+ objects = set()
+ if shared:
+ path, node = self._lookup_container(account, container)
+ shared = self._get_formatted_paths(shared)
+ objects |= set(self._list_object_properties(node, path, prefix, delimiter, marker, limit, virtual, domain, keys, until, size_range, shared, all_props))
+
+ # get public
+ objects |= set(self._list_public_object_properties(user, account, container, prefix, all_props))
+ objects = list(objects)
+
+ objects.sort(key=lambda x: x[0])
+ start, limit = self._list_limits([x[0] for x in objects], marker, limit)
+ return objects[start:start + limit]
+ elif public:
+ objects = self._list_public_object_properties(user, account, container, prefix, all_props)
+ start, limit = self._list_limits([x[0] for x in objects], marker, limit)
+ return objects[start:start + limit]
+
allowed = self._list_object_permissions(user, account, container, prefix, shared, public)
- if (shared or public) and not allowed:
+ if shared and not allowed:
return []
path, node = self._lookup_container(account, container)
allowed = self._get_formatted_paths(allowed)
- return self._list_object_properties(node, path, prefix, delimiter, marker, limit, virtual, domain, keys, until, size_range, allowed, all_props)
+ objects = self._list_object_properties(node, path, prefix, delimiter, marker, limit, virtual, domain, keys, until, size_range, allowed, all_props)
+ start, limit = self._list_limits([x[0] for x in objects], marker, limit)
+ return objects[start:start + limit]
+
+ def _list_public_object_properties(self, user, account, container, prefix, all_props):
+ public = self._list_object_permissions(user, account, container, prefix, shared=False, public=True)
+ paths, nodes = self._lookup_objects(public)
+ path = '/'.join((account, container))
+ cont_prefix = path + '/'
+ paths = [x[len(cont_prefix):] for x in paths]
+ props = self.node.version_lookup_bulk(nodes, all_props=all_props)
+ objects = [(path,) + props for path, props in zip(paths, props)]
+ return objects
+
+ def _list_objects_no_limit(self, user, account, container, prefix, delimiter, virtual, domain, keys, shared, until, size_range, all_props, public):
+ objects = []
+ while True:
+ marker = objects[-1] if objects else None
+ limit = 10000
+ l = self._list_objects(user, account, container, prefix, delimiter, marker, limit, virtual, domain, keys, shared, until, size_range, all_props, public)
+ objects.extend(l)
+ if not l or len(l) < limit:
+ break
+ return objects
def _list_object_permissions(self, user, account, container, prefix, shared, public):
allowed = []
if not allowed:
raise NotAllowedError
else:
- allowed = []
+ allowed = set()
if shared:
- allowed.extend(self.permissions.access_list_shared(path))
+ allowed.update(self.permissions.access_list_shared(path))
if public:
- allowed.extend([x[0] for x in self.permissions.public_list(path)])
- allowed = list(set(allowed))
- allowed.sort()
+ allowed.update([x[0] for x in self.permissions.public_list(path)])
+ allowed = sorted(allowed)
if not allowed:
return []
return allowed
except NameError: # Object may be deleted.
del_props = self.node.version_lookup(node, inf, CLUSTER_DELETED)
if del_props is None:
- raise NameError('Object does not exist')
+ raise ItemNotExists('Object does not exist')
modified = del_props[self.MTIME]
meta = {}
(container_quota > 0 and self._get_statistics(container_node)[1] + size_delta > container_quota):
# This must be executed in a transaction, so the version is never created if it fails.
raise QuotaError
- self._report_size_change(user, account, size_delta, {'action': 'object update'})
+ self._report_size_change(user, account, size_delta, {'action': 'object update', 'path':path})
if permissions is not None:
self.permissions.access_set(path, permissions)
if x[self.SERIAL] >= int(version) and x[self.HASH] == props[self.HASH] and x[self.SIZE] == props[self.SIZE]:
self.node.version_put_property(x[self.SERIAL], 'checksum', checksum)
- def _copy_object(self, user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, dest_domain=None, dest_meta={}, replace_meta=False, permissions=None, src_version=None, is_move=False):
+ def _copy_object(self, user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, dest_domain=None, dest_meta={}, replace_meta=False, permissions=None, src_version=None, is_move=False, delimiter=None):
+ dest_version_ids = []
self._can_read(user, src_account, src_container, src_name)
path, node = self._lookup_object(src_account, src_container, src_name)
# TODO: Will do another fetch of the properties in duplicate version...
src_version_id = props[self.SERIAL]
hash = props[self.HASH]
size = props[self.SIZE]
-
is_copy = not is_move and (src_account, src_container, src_name) != (dest_account, dest_container, dest_name) # New uuid.
- dest_version_id = self._update_object_hash(user, dest_account, dest_container, dest_name, size, type, hash, None, dest_domain, dest_meta, replace_meta, permissions, src_node=node, src_version_id=src_version_id, is_copy=is_copy)
- return dest_version_id
+ dest_version_ids.append(self._update_object_hash(user, dest_account, dest_container, dest_name, size, type, hash, None, dest_domain, dest_meta, replace_meta, permissions, src_node=node, src_version_id=src_version_id, is_copy=is_copy))
+ if is_move and (src_account, src_container, src_name) != (dest_account, dest_container, dest_name):
+ self._delete_object(user, src_account, src_container, src_name)
+
+ if delimiter:
+ prefix = src_name + delimiter if not src_name.endswith(delimiter) else src_name
+ src_names = self._list_objects_no_limit(user, src_account, src_container, prefix, delimiter=None, virtual=False, domain=None, keys=[], shared=False, until=None, size_range=None, all_props=True, public=False)
+ src_names.sort(key=lambda x: x[2]) # order by nodes
+ paths = [elem[0] for elem in src_names]
+ nodes = [elem[2] for elem in src_names]
+ # TODO: Will do another fetch of the properties in duplicate version...
+ props = self._get_versions(nodes) # Check to see if source exists.
+
+ for prop, path, node in zip(props, paths, nodes):
+ src_version_id = prop[self.SERIAL]
+ hash = prop[self.HASH]
+ vtype = prop[self.TYPE]
+ size = prop[self.SIZE]
+ dest_prefix = dest_name + delimiter if not dest_name.endswith(delimiter) else dest_name
+ vdest_name = path.replace(prefix, dest_prefix, 1)
+ dest_version_ids.append(self._update_object_hash(user, dest_account, dest_container, vdest_name, size, vtype, hash, None, dest_domain, meta={}, replace_meta=False, permissions=None, src_node=node, src_version_id=src_version_id, is_copy=is_copy))
+ if is_move and (src_account, src_container, src_name) != (dest_account, dest_container, dest_name):
+ self._delete_object(user, src_account, src_container, path)
+ return dest_version_ids[0] if len(dest_version_ids) == 1 else dest_version_ids
@backend_method
- def copy_object(self, user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta={}, replace_meta=False, permissions=None, src_version=None):
+ def copy_object(self, user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta={}, replace_meta=False, permissions=None, src_version=None, delimiter=None):
"""Copy an object's data and metadata."""
- logger.debug("copy_object: %s %s %s %s %s %s %s %s %s %s %s %s %s", user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, src_version)
- dest_version_id = self._copy_object(user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, src_version, False)
+ logger.debug("copy_object: %s %s %s %s %s %s %s %s %s %s %s %s %s %s", user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, src_version, delimiter)
+ dest_version_id = self._copy_object(user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, src_version, False, delimiter)
return dest_version_id
@backend_method
- def move_object(self, user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta={}, replace_meta=False, permissions=None):
+ def move_object(self, user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta={}, replace_meta=False, permissions=None, delimiter=None):
"""Move an object's data and metadata."""
- logger.debug("move_object: %s %s %s %s %s %s %s %s %s %s %s %s", user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions)
+ logger.debug("move_object: %s %s %s %s %s %s %s %s %s %s %s %s %s", user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, delimiter)
if user != src_account:
raise NotAllowedError
- dest_version_id = self._copy_object(user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, None, True)
- if (src_account, src_container, src_name) != (dest_account, dest_container, dest_name):
- self._delete_object(user, src_account, src_container, src_name)
+ dest_version_id = self._copy_object(user, src_account, src_container, src_name, dest_account, dest_container, dest_name, type, domain, meta, replace_meta, permissions, None, True, delimiter)
return dest_version_id
- def _delete_object(self, user, account, container, name, until=None, prefix='', delimiter=None):
+ def _delete_object(self, user, account, container, name, until=None, delimiter=None):
if user != account:
raise NotAllowedError
props = self._get_version(node)
except NameError:
self.permissions.access_clear(path)
- self._report_size_change(user, account, -size, {'action': 'object purge'})
+ self._report_size_change(user, account, -size, {'action': 'object purge', 'path':path})
return
path, node = self._lookup_object(account, container, name)
src_version_id, dest_version_id = self._put_version_duplicate(user, node, size=0, type='', hash=None, checksum='', cluster=CLUSTER_DELETED)
del_size = self._apply_versioning(account, container, src_version_id)
if del_size:
- self._report_size_change(user, account, -del_size, {'action': 'object delete'})
+ self._report_size_change(user, account, -del_size, {'action': 'object delete', 'path':path})
self._report_object_change(user, account, path, details={'action': 'object delete'})
self.permissions.access_clear(path)
+
+ if delimiter:
+ prefix = name + delimiter if not name.endswith(delimiter) else name
+ src_names = self._list_objects_no_limit(user, account, container, prefix, delimiter=None, virtual=False, domain=None, keys=[], shared=False, until=None, size_range=None, all_props=True, public=False)
+ paths = []
+ for t in src_names:
+ path = '/'.join((account, container, t[0]))
+ node = t[2]
+ src_version_id, dest_version_id = self._put_version_duplicate(user, node, size=0, type='', hash=None, checksum='', cluster=CLUSTER_DELETED)
+ del_size = self._apply_versioning(account, container, src_version_id)
+ if del_size:
+ self._report_size_change(user, account, -del_size, {'action': 'object delete', 'path':path})
+ self._report_object_change(user, account, path, details={'action': 'object delete'})
+ paths.append(path)
+ self.permissions.access_clear_bulk(paths)
@backend_method
- def (self, user, account, container, name, until=None, prefix='', delimiter=None):
+ def delete_object(self, user, account, container, name, until=None, prefix='', delimiter=None):
"""Delete/purge an object."""
logger.debug("delete_object: %s %s %s %s %s %s %s", user, account, container, name, until, prefix, delimiter)
- self._delete_object(user, account, container, name, until)
+ self._delete_object(user, account, container, name, until, delimiter)
@backend_method
def list_versions(self, user, account, container, name):
logger.debug("get_block: %s", hash)
block = self.store.block_get(binascii.unhexlify(hash))
if not block:
- raise NameError('Block does not exist')
+ raise ItemNotExists('Block does not exist')
return block
@backend_method(autocommit=0)
path = '/'.join((account, container))
node = self.node.node_lookup(path)
if node is None:
- raise NameError('Container does not exist')
+ raise ItemNotExists('Container does not exist')
return path, node
def _lookup_object(self, account, container, name):
path = '/'.join((account, container, name))
node = self.node.node_lookup(path)
if node is None:
- raise NameError('Object does not exist')
+ raise ItemNotExists('Object does not exist')
return path, node
+ def _lookup_objects(self, paths):
+ nodes = self.node.node_lookup_bulk(paths)
+ return paths, nodes
+
def _get_properties(self, node, until=None):
"""Return properties until the timestamp given."""
if props is None and until is not None:
props = self.node.version_lookup(node, before, CLUSTER_HISTORY)
if props is None:
- raise NameError('Path does not exist')
+ raise ItemNotExists('Path does not exist')
return props
def _get_statistics(self, node, until=None):
if version is None:
props = self.node.version_lookup(node, inf, CLUSTER_NORMAL)
if props is None:
- raise NameError('Object does not exist')
+ raise ItemNotExists('Object does not exist')
else:
try:
version = int(version)
except ValueError:
- raise IndexError('Version does not exist')
+ raise VersionNotExists('Version does not exist')
props = self.node.version_get_properties(version)
if props is None or props[self.CLUSTER] == CLUSTER_DELETED:
- raise IndexError('Version does not exist')
+ raise VersionNotExists('Version does not exist')
return props
+
+ def _get_versions(self, nodes):
+ return self.node.version_lookup_bulk(nodes, inf, CLUSTER_NORMAL)
def _put_version_duplicate(self, user, node, src_node=None, size=None, type=None, hash=None, checksum=None, cluster=CLUSTER_NORMAL, is_copy=False):
"""Create a new version of the node."""
objects.extend([(p, None) for p in prefixes] if virtual else [])
objects.sort(key=lambda x: x[0])
objects = [(x[0][len(cont_prefix):],) + x[1:] for x in objects]
+ return objects
- start, limit = self._list_limits([x[0] for x in objects], marker, limit)
- return objects[start:start + limit]
-
# Reporting functions.
def _report_size_change(self, user, account, size, details={}):
- logger.debug("_report_size_change: %s %s %s %s", user, account, size, details)
account_node = self._lookup_account(account, True)[1]
total = self._get_statistics(account_node)[1]
details.update({'user': user, 'total': total})
+ logger.debug("_report_size_change: %s %s %s %s", user, account, size, details)
self.messages.append((QUEUE_MESSAGE_KEY_PREFIX % ('resource.diskspace',), account, QUEUE_INSTANCE_ID, 'diskspace', float(size), details))
def _report_object_change(self, user, account, path, details={}):
- logger.debug("_report_object_change: %s %s %s %s", user, account, path, details)
details.update({'user': user})
+ logger.debug("_report_object_change: %s %s %s %s", user, account, path, details)
self.messages.append((QUEUE_MESSAGE_KEY_PREFIX % ('object',), account, QUEUE_INSTANCE_ID, 'object', path, details))
def _report_sharing_change(self, user, account, path, details={}):
projects / pithos / blobdiff