X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/86f94e641462ddd78a454fc0a0cdc9dd4c4d9bef..cef69b16e1295e456548813ad95c9db827136c17:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index ea18123f65..20f5c40bfc 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -3,17 +3,21 @@ import logging import os import re import errno +import time from collections import deque from stat import * -from .arvfile import ArvadosFileBase, split, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, BlockManager +from .arvfile import ArvadosFileBase, split, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, BlockManager, synchronized, must_be_writable, SYNC_READONLY, SYNC_EXPLICIT, SYNC_LIVE, NoopLock from keep import * from .stream import StreamReader, normalize_stream, locator_block_size from .ranges import Range, LocatorAndRange +from .safeapi import SafeApi import config import errors import util +import events +from arvados.retry import retry_method _logger = logging.getLogger('arvados.collection') @@ -639,22 +643,517 @@ class ResumableCollectionWriter(CollectionWriter): "resumable writer can't accept unsourced data") return super(ResumableCollectionWriter, self).write(data) +ADD = "add" +DEL = "del" +MOD = "mod" -class Collection(CollectionBase): - def __init__(self, manifest_locator_or_text=None, parent=None, api_client=None, - keep_client=None, num_retries=0, block_manager=None): +class SynchronizedCollectionBase(CollectionBase): + """Base class for Collections and Subcollections. Implements the majority of + functionality relating to accessing items in the Collection.""" + def __init__(self, parent=None): self.parent = parent - self._items = None + self._modified = True + self._items = {} + + def _my_api(self): + raise NotImplementedError() + + def _my_keep(self): + raise NotImplementedError() + + def _my_block_manager(self): + raise NotImplementedError() + + def _populate(self): + raise NotImplementedError() + + def sync_mode(self): + raise NotImplementedError() + + def root_collection(self): + raise NotImplementedError() + + def notify(self, event, collection, name, item): + raise NotImplementedError() + + @synchronized + def find(self, path, create=False, create_collection=False): + """Recursively search the specified file path. May return either a Collection + or ArvadosFile. + + :create: + If true, create path components (i.e. Collections) that are + missing. If "create" is False, return None if a path component is + not found. + + :create_collection: + If the path is not found, "create" is True, and + "create_collection" is False, then create and return a new + ArvadosFile for the last path component. If "create_collection" is + True, then create and return a new Collection for the last path + component. + + """ + if create and self.sync_mode() == SYNC_READONLY: + raise IOError((errno.EROFS, "Collection is read only")) + + p = path.split("/") + if p[0] == '.': + del p[0] + + if p and p[0]: + item = self._items.get(p[0]) + if len(p) == 1: + # item must be a file + if item is None and create: + # create new file + if create_collection: + item = Subcollection(self) + else: + item = ArvadosFile(self) + self._items[p[0]] = item + self._modified = True + self.notify(ADD, self, p[0], item) + return item + else: + if item is None and create: + # create new collection + item = Subcollection(self) + self._items[p[0]] = item + self._modified = True + self.notify(ADD, self, p[0], item) + del p[0] + if isinstance(item, SynchronizedCollectionBase): + return item.find("/".join(p), create=create) + else: + raise errors.ArgumentError("Interior path components must be subcollection") + else: + return self + + def open(self, path, mode): + """Open a file-like object for access. + + :path: + path to a file in the collection + :mode: + one of "r", "r+", "w", "w+", "a", "a+" + :"r": + opens for reading + :"r+": + opens for reading and writing. Reads/writes share a file pointer. + :"w", "w+": + truncates to 0 and opens for reading and writing. Reads/writes share a file pointer. + :"a", "a+": + opens for reading and writing. All writes are appended to + the end of the file. Writing does not affect the file pointer for + reading. + """ + mode = mode.replace("b", "") + if len(mode) == 0 or mode[0] not in ("r", "w", "a"): + raise ArgumentError("Bad mode '%s'" % mode) + create = (mode != "r") + + if create and self.sync_mode() == SYNC_READONLY: + raise IOError((errno.EROFS, "Collection is read only")) + + f = self.find(path, create=create) + + if f is None: + raise IOError((errno.ENOENT, "File not found")) + if not isinstance(f, ArvadosFile): + raise IOError((errno.EISDIR, "Path must refer to a file.")) + + if mode[0] == "w": + f.truncate(0) + + if mode == "r": + return ArvadosFileReader(f, path, mode, num_retries=self.num_retries) + else: + return ArvadosFileWriter(f, path, mode, num_retries=self.num_retries) + + @synchronized + def modified(self): + """Test if the collection (or any subcollection or file) has been modified + since it was created.""" + if self._modified: + return True + for k,v in self._items.items(): + if v.modified(): + return True + return False + + @synchronized + def set_unmodified(self): + """Recursively clear modified flag""" + self._modified = False + for k,v in self._items.items(): + v.set_unmodified() + + @synchronized + def __iter__(self): + """Iterate over names of files and collections contained in this collection.""" + return self._items.keys().__iter__() + + @synchronized + def iterkeys(self): + """Iterate over names of files and collections directly contained in this collection.""" + return self._items.keys() + + @synchronized + def __getitem__(self, k): + """Get a file or collection that is directly contained by this collection. If + you want to search a path, use `find()` instead. + """ + return self._items[k] + + @synchronized + def __contains__(self, k): + """If there is a file or collection a directly contained by this collection + with name "k".""" + return k in self._items + + @synchronized + def __len__(self): + """Get the number of items directly contained in this collection""" + return len(self._items) + + @must_be_writable + @synchronized + def __delitem__(self, p): + """Delete an item by name which is directly contained by this collection.""" + del self._items[p] + self._modified = True + self.notify(DEL, self, p, None) + + @synchronized + def keys(self): + """Get a list of names of files and collections directly contained in this collection.""" + return self._items.keys() + + @synchronized + def values(self): + """Get a list of files and collection objects directly contained in this collection.""" + return self._items.values() + + @synchronized + def items(self): + """Get a list of (name, object) tuples directly contained in this collection.""" + return self._items.items() + + def exists(self, path): + """Test if there is a file or collection at "path" """ + return self.find(path) != None + + @must_be_writable + @synchronized + def remove(self, path, rm_r=False): + """Remove the file or subcollection (directory) at `path`. + :rm_r: + Specify whether to remove non-empty subcollections (True), or raise an error (False). + """ + p = path.split("/") + if p[0] == '.': + # Remove '.' from the front of the path + del p[0] + + if len(p) > 0: + item = self._items.get(p[0]) + if item is None: + raise IOError((errno.ENOENT, "File not found")) + if len(p) == 1: + if isinstance(self._items[p[0]], SynchronizedCollectionBase) and len(self._items[p[0]]) > 0 and not rm_r: + raise IOError((errno.ENOTEMPTY, "Subcollection not empty")) + d = self._items[p[0]] + del self._items[p[0]] + self._modified = True + self.notify(DEL, self, p[0], d) + else: + del p[0] + item.remove("/".join(p)) + else: + raise IOError((errno.ENOENT, "File not found")) + + def _cloneinto(self, target): + for k,v in self._items.items(): + target._items[k] = v.clone(target) + + def clone(self): + raise NotImplementedError() + + @must_be_writable + @synchronized + def copy(self, source, target_path, source_collection=None, overwrite=False): + """Copy a file or subcollection to a new path in this collection. + + :source: + An ArvadosFile, Subcollection, or string with a path to source file or subcollection + + :target_path: + Destination file or path. If the target path already exists and is a + subcollection, the item will be placed inside the subcollection. If + the target path already exists and is a file, this will raise an error + unless you specify `overwrite=True`. + + :source_collection: + Collection to copy `source_path` from (default `self`) + + :overwrite: + Whether to overwrite target file if it already exists. + """ + if source_collection is None: + source_collection = self + + # Find the object to copy + if isinstance(source, basestring): + source_obj = source_collection.find(source) + if source_obj is None: + raise IOError((errno.ENOENT, "File not found")) + sp = source.split("/") + else: + source_obj = source + sp = None + + # Find parent collection the target path + tp = target_path.split("/") + + # Determine the name to use. + target_name = tp[-1] if tp[-1] else (sp[-1] if sp else None) + + if not target_name: + raise errors.ArgumentError("Target path is empty and source is an object. Cannot determine destination filename to use.") + + target_dir = self.find("/".join(tp[0:-1]), create=True, create_collection=True) + + with target_dir.lock: + if target_name in target_dir: + if isinstance(target_dir[target_name], SynchronizedCollectionBase) and sp: + target_dir = target_dir[target_name] + target_name = sp[-1] + elif not overwrite: + raise IOError((errno.EEXIST, "File already exists")) + + mod = None + if target_name in target_dir: + mod = target_dir[target_name] + + # Actually make the copy. + dup = source_obj.clone(target_dir) + target_dir._items[target_name] = dup + target_dir._modified = True + + if mod: + self.notify(MOD, target_dir, target_name, (mod, dup)) + else: + self.notify(ADD, target_dir, target_name, dup) + + @synchronized + def manifest_text(self, strip=False, normalize=False): + """Get the manifest text for this collection, sub collections and files. + + :strip: + If True, remove signing tokens from block locators if present. + If False, block locators are left unchanged. + + :normalize: + If True, always export the manifest text in normalized form + even if the Collection is not modified. If False and the collection + is not modified, return the original manifest text even if it is not + in normalized form. + + """ + if self.modified() or self._manifest_text is None or normalize: + return export_manifest(self, stream_name=".", portable_locators=strip) + else: + if strip: + return self.stripped_manifest() + else: + return self._manifest_text + + @synchronized + def diff(self, end_collection, prefix=".", holding_collection=None): + """ + Generate list of add/modify/delete actions which, when given to `apply`, will + change `self` to match `end_collection` + """ + changes = [] + if holding_collection is None: + holding_collection = CollectionRoot(api_client=self._my_api(), keep_client=self._my_keep(), sync=SYNC_READONLY) + for k in self: + if k not in end_collection: + changes.append((DEL, os.path.join(prefix, k), self[k].clone(holding_collection))) + for k in end_collection: + if k in self: + if isinstance(end_collection[k], Subcollection) and isinstance(self[k], Subcollection): + changes.extend(self[k].diff(end_collection[k], os.path.join(prefix, k), holding_collection)) + elif end_collection[k] != self[k]: + changes.append((MOD, os.path.join(prefix, k), self[k].clone(holding_collection), end_collection[k].clone(holding_collection))) + else: + changes.append((ADD, os.path.join(prefix, k), end_collection[k].clone(holding_collection))) + return changes + + @must_be_writable + @synchronized + def apply(self, changes): + """ + Apply changes from `diff`. If a change conflicts with a local change, it + will be saved to an alternate path indicating the conflict. + """ + for c in changes: + path = c[1] + initial = c[2] + local = self.find(path) + conflictpath = "%s~conflict-%s~" % (path, time.strftime("%Y-%m-%d-%H:%M:%S", + time.gmtime())) + if c[0] == ADD: + if local is None: + # No local file at path, safe to copy over new file + self.copy(initial, path) + elif local is not None and local != initial: + # There is already local file and it is different: + # save change to conflict file. + self.copy(initial, conflictpath) + elif c[0] == MOD: + if local == initial: + # Local matches the "initial" item so it has not + # changed locally and is safe to update. + if isinstance(local, ArvadosFile) and isinstance(c[3], ArvadosFile): + # Replace contents of local file with new contents + local.replace_contents(c[3]) + else: + # Overwrite path with new item; this can happen if + # path was a file and is now a collection or vice versa + self.copy(c[3], path, overwrite=True) + else: + # Local is missing (presumably deleted) or local doesn't + # match the "start" value, so save change to conflict file + self.copy(c[3], conflictpath) + elif c[0] == DEL: + if local == initial: + # Local item matches "initial" value, so it is safe to remove. + self.remove(path, rm_r=True) + # else, the file is modified or already removed, in either + # case we don't want to try to remove it. + + def portable_data_hash(self): + """Get the portable data hash for this collection's manifest.""" + stripped = self.manifest_text(strip=True) + return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped)) + + @synchronized + def __eq__(self, other): + if other is self: + return True + if not isinstance(other, SynchronizedCollectionBase): + return False + if len(self._items) != len(other): + return False + for k in self._items: + if k not in other: + return False + if self._items[k] != other[k]: + return False + return True + + def __ne__(self, other): + return not self.__eq__(other) + +class CollectionRoot(SynchronizedCollectionBase): + """Represents the root of an Arvados Collection, which may be associated with + an API server Collection record. + + Brief summary of useful methods: + + :To read an existing file: + `c.open("myfile", "r")` + + :To write a new file: + `c.open("myfile", "w")` + + :To determine if a file exists: + `c.find("myfile") is not None` + + :To copy a file: + `c.copy("source", "dest")` + + :To delete a file: + `c.remove("myfile")` + + :To save to an existing collection record: + `c.save()` + + :To save a new collection record: + `c.save_new()` + + :To merge remote changes into this object: + `c.update()` + + This class is threadsafe. The root collection object, all subcollections + and files are protected by a single lock (i.e. each access locks the entire + collection). + + """ + + def __init__(self, manifest_locator_or_text=None, + parent=None, + apiconfig=None, + api_client=None, + keep_client=None, + num_retries=None, + block_manager=None, + sync=None): + """:manifest_locator_or_text: + One of Arvados collection UUID, block locator of + a manifest, raw manifest text, or None (to create an empty collection). + :parent: + the parent Collection, may be None. + :apiconfig: + A dict containing keys for ARVADOS_API_HOST and ARVADOS_API_TOKEN. + Prefer this over supplying your own api_client and keep_client (except in testing). + Will use default config settings if not specified. + :api_client: + The API client object to use for requests. If not specified, create one using `apiconfig`. + :keep_client: + the Keep client to use for requests. If not specified, create one using `apiconfig`. + :num_retries: + the number of retries for API and Keep requests. + :block_manager: + the block manager to use. If not specified, create one. + :sync: + Set synchronization policy with API server collection record. + :SYNC_READONLY: + Collection is read only. No synchronization. This mode will + also forego locking, which gives better performance. + :SYNC_EXPLICIT: + Collection is writable. Synchronize on explicit request via `update()` or `save()` + :SYNC_LIVE: + Collection is writable. Synchronize with server in response to + background websocket events, on block write, or on file close. + + """ + super(CollectionRoot, self).__init__(parent) self._api_client = api_client self._keep_client = keep_client self._block_manager = block_manager + if apiconfig: + self._config = apiconfig + else: + self._config = config.settings() + self.num_retries = num_retries self._manifest_locator = None self._manifest_text = None self._api_response = None + if sync is None: + raise errors.ArgumentError("Must specify sync mode") + + self._sync = sync + self.lock = threading.RLock() + self.callbacks = [] + self.events = None + if manifest_locator_or_text: if re.match(util.keep_locator_pattern, manifest_locator_or_text): self._manifest_locator = manifest_locator_or_text @@ -666,26 +1165,65 @@ class Collection(CollectionBase): raise errors.ArgumentError( "Argument to CollectionReader must be a manifest or a collection UUID") + self._populate() + + if self._sync == SYNC_LIVE: + if not self._has_collection_uuid(): + raise errors.ArgumentError("Cannot SYNC_LIVE associated with a collection uuid") + self.events = events.subscribe(arvados.api(apiconfig=self._config), + [["object_uuid", "=", self._manifest_locator]], + self.on_message) + + + def root_collection(self): + return self + + def sync_mode(self): + return self._sync + + def on_message(self, event): + if event.get("object_uuid") == self._manifest_locator: + self.update() + + @staticmethod + def create(name, owner_uuid=None, sync=SYNC_EXPLICIT, apiconfig=None): + """Create a new empty Collection with associated collection record.""" + c = Collection(sync=SYNC_EXPLICIT, apiconfig=apiconfig) + c.save_new(name, owner_uuid=owner_uuid, ensure_unique_name=True) + if sync == SYNC_LIVE: + c.events = events.subscribe(arvados.api(apiconfig=self._config), [["object_uuid", "=", c._manifest_locator]], c.on_message) + return c + + @synchronized + @retry_method + def update(self, other=None, num_retries=None): + if other is None: + if self._manifest_locator is None: + raise errors.ArgumentError("`other` is None but collection does not have a manifest_locator uuid") + n = self._my_api().collections().get(uuid=self._manifest_locator).execute(num_retries=num_retries) + other = import_collection(n["manifest_text"]) + baseline = import_collection(self._manifest_text) + self.apply(other.diff(baseline)) + + @synchronized def _my_api(self): if self._api_client is None: - if self.parent is not None: - return self.parent._my_api() - self._api_client = arvados.api('v1') - self._keep_client = None # Make a new one with the new api. + self._api_client = arvados.SafeApi(self._config) + self._keep_client = self._api_client.keep return self._api_client + @synchronized def _my_keep(self): if self._keep_client is None: - if self.parent is not None: - return self.parent._my_keep() - self._keep_client = KeepClient(api_client=self._my_api(), - num_retries=self.num_retries) + if self._api_client is None: + self._my_api() + else: + self._keep_client = KeepClient(api=self._api_client) return self._keep_client + @synchronized def _my_block_manager(self): if self._block_manager is None: - if self.parent is not None: - return self.parent._my_block_manager() self._block_manager = BlockManager(self._my_keep()) return self._block_manager @@ -718,14 +1256,13 @@ class Collection(CollectionBase): return e def _populate(self): - self._items = {} if self._manifest_locator is None and self._manifest_text is None: return error_via_api = None error_via_keep = None should_try_keep = ((self._manifest_text is None) and util.keep_locator_pattern.match( - self._manifest_locator)) + self._manifest_locator)) if ((self._manifest_text is None) and util.signed_locator_pattern.match(self._manifest_locator)): error_via_keep = self._populate_from_keep() @@ -748,56 +1285,42 @@ class Collection(CollectionBase): error_via_api, error_via_keep)) # populate + self._baseline_manifest = self._manifest_text import_manifest(self._manifest_text, self) - def _populate_first(orig_func): - # Decorator for methods that read actual Collection data. - @functools.wraps(orig_func) - def wrapper(self, *args, **kwargs): - if self._items is None: - self._populate() - return orig_func(self, *args, **kwargs) - return wrapper + if self._sync == SYNC_READONLY: + # Now that we're populated, knowing that this will be readonly, + # forego any further locking. + self.lock = NoopLock() + + def _has_collection_uuid(self): + return self._manifest_locator is not None and re.match(util.collection_uuid_pattern, self._manifest_locator) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): - self.save(no_locator=True) + """Support scoped auto-commit in a with: block""" + if self._sync != SYNC_READONLY and self._has_collection_uuid(): + self.save() if self._block_manager is not None: self._block_manager.stop_threads() - @_populate_first - def find(self, path, create=False, create_collection=False): - p = path.split("/") - if p[0] == '.': - del p[0] - - if len(p) > 0: - item = self._items.get(p[0]) - if len(p) == 1: - # item must be a file - if item is None and create: - # create new file - if create_collection: - item = Collection(parent=self, num_retries=self.num_retries) - else: - item = ArvadosFile(self) - self._items[p[0]] = item - return item - else: - if item is None and create: - # create new collection - item = Collection(parent=self, num_retries=self.num_retries) - self._items[p[0]] = item - del p[0] - return item.find("/".join(p), create=create) - else: - return self - - @_populate_first + @synchronized + def clone(self, new_parent=None, new_sync=SYNC_READONLY, new_config=None): + if new_config is None: + new_config = self._config + c = CollectionRoot(parent=new_parent, apiconfig=new_config, sync=new_sync) + if new_sync == SYNC_READONLY: + c.lock = NoopLock() + c._items = {} + self._cloneinto(c) + return c + + @synchronized def api_response(self): - """api_response() -> dict or None + """ + api_response() -> dict or None Returns information about this Collection fetched from the API server. If the Collection exists in Keep but not the API server, currently @@ -805,164 +1328,186 @@ class Collection(CollectionBase): """ return self._api_response - def open(self, path, mode): - mode = mode.replace("b", "") - if len(mode) == 0 or mode[0] not in ("r", "w", "a"): - raise ArgumentError("Bad mode '%s'" % mode) - create = (mode != "r") + @must_be_writable + @synchronized + @retry_method + def save(self, merge=True, num_retries=None): + """Commit pending buffer blocks to Keep, merge with remote record (if + update=True), write the manifest to Keep, and update the collection + record. Will raise AssertionError if not associated with a collection + record on the API server. If you want to save a manifest to Keep only, + see `save_new()`. - f = self.find(path, create=create) - if f is None: - raise IOError((errno.ENOENT, "File not found")) - if not isinstance(f, ArvadosFile): - raise IOError((errno.EISDIR, "Path must refer to a file.")) + :update: + Update and merge remote changes before saving. Otherwise, any + remote changes will be ignored and overwritten. - if mode[0] == "w": - f.truncate(0) + """ + if self.modified(): + if not self._has_collection_uuid(): + raise AssertionError("Collection manifest_locator must be a collection uuid. Use save_as() for new collections.") + self._my_block_manager().commit_all() + if merge: + self.update() + self._my_keep().put(self.manifest_text(strip=True), num_retries=num_retries) + + mt = self.manifest_text(strip=False) + self._api_response = self._my_api().collections().update( + uuid=self._manifest_locator, + body={'manifest_text': mt} + ).execute( + num_retries=num_retries) + self._manifest_text = mt + self.set_unmodified() - if mode == "r": - return ArvadosFileReader(f, path, mode) - else: - return ArvadosFileWriter(f, path, mode) + @must_be_writable + @synchronized + @retry_method + def save_new(self, name=None, create_collection_record=True, owner_uuid=None, ensure_unique_name=False, num_retries=None): + """Commit pending buffer blocks to Keep, write the manifest to Keep, and create + a new collection record (if create_collection_record True). After + creating a new collection record, this Collection object will be + associated with the new record for `save()` and SYNC_LIVE updates. - @_populate_first - def modified(self): - for k,v in self._items.items(): - if v.modified(): - return True - return False + :name: + The collection name. - @_populate_first - def set_unmodified(self): - for k,v in self._items.items(): - v.set_unmodified() + :keep_only: + Only save the manifest to keep, do not create a collection record. - @_populate_first - def __iter__(self): - return self._items.iterkeys() + :owner_uuid: + the user, or project uuid that will own this collection. + If None, defaults to the current user. - @_populate_first - def iterkeys(self): - return self._items.iterkeys() + :ensure_unique_name: + If True, ask the API server to rename the collection + if it conflicts with a collection with the same name and owner. If + False, a name conflict will result in an error. - @_populate_first - def __getitem__(self, k): - return self._items[k] + """ + self._my_block_manager().commit_all() + self._my_keep().put(self.manifest_text(strip=True), num_retries=num_retries) + mt = self.manifest_text(strip=False) - @_populate_first - def __contains__(self, k): - return k in self._items + if create_collection_record: + if name is None: + name = "Collection created %s" % (time.strftime("%Y-%m-%d %H:%M:%S %Z", time.localtime())) - @_populate_first - def __len__(self): - return len(self._items) + body = {"manifest_text": mt, + "name": name} + if owner_uuid: + body["owner_uuid"] = owner_uuid - @_populate_first - def __delitem__(self, p): - del self._items[p] + self._api_response = self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=num_retries) - @_populate_first - def keys(self): - return self._items.keys() + if self.events: + self.events.unsubscribe(filters=[["object_uuid", "=", self._manifest_locator]]) - @_populate_first - def values(self): - return self._items.values() + self._manifest_locator = self._api_response["uuid"] - @_populate_first - def items(self): - return self._items.items() + if self.events: + self.events.subscribe(filters=[["object_uuid", "=", self._manifest_locator]]) - @_populate_first - def exists(self, path): - return self.find(path) != None + self._manifest_text = mt + self.set_unmodified() - @_populate_first - def remove(self, path): - p = path.split("/") - if p[0] == '.': - del p[0] + @synchronized + def subscribe(self, callback): + self.callbacks.append(callback) - if len(p) > 0: - item = self._items.get(p[0]) - if item is None: - raise IOError((errno.ENOENT, "File not found")) - if len(p) == 1: - del self._items[p[0]] - else: - del p[0] - item.remove("/".join(p)) - else: - raise IOError((errno.ENOENT, "File not found")) + @synchronized + def unsubscribe(self, callback): + self.callbacks.remove(callback) - @_populate_first - def manifest_text(self, strip=False, normalize=False): - if self.modified() or self._manifest_text is None or normalize: - return export_manifest(self, stream_name=".", portable_locators=strip) - else: - if strip: - return self.stripped_manifest() - else: - return self._manifest_text + @synchronized + def notify(self, event, collection, name, item): + for c in self.callbacks: + c(event, collection, name, item) - def portable_data_hash(self): - stripped = self.manifest_text(strip=True) - return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped)) +def ReadOnlyCollection(*args, **kwargs): + kwargs["sync"] = SYNC_READONLY + return CollectionRoot(*args, **kwargs) - @_populate_first - def save(self, no_locator=False): - if self.modified(): - self._my_block_manager().commit_all() - self._my_keep().put(self.manifest_text(strip=True)) - if self._manifest_locator is not None and re.match(util.collection_uuid_pattern, self._manifest_locator): - self._api_response = self._my_api().collections().update( - uuid=self._manifest_locator, - body={'manifest_text': self.manifest_text(strip=False)} - ).execute( - num_retries=self.num_retries) - elif not no_locator: - raise AssertionError("Collection manifest_locator must be a collection uuid. Use save_as() for new collections.") - self.set_unmodified() +def WritableCollection(*args, **kwargs): + kwargs["sync"] = SYNC_EXPLICIT + return CollectionRoot(*args, **kwargs) - @_populate_first - def save_as(self, name, owner_uuid=None, ensure_unique_name=False): - self._my_block_manager().commit_all() - self._my_keep().put(self.manifest_text(strip=True)) - body = {"manifest_text": self.manifest_text(strip=False), - "name": name} - if owner_uuid: - body["owner_uuid"] = owner_uuid - self._api_response = self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=self.num_retries) - self._manifest_locator = self._api_response["uuid"] - self.set_unmodified() +def LiveCollection(*args, **kwargs): + kwargs["sync"] = SYNC_LIVE + return CollectionRoot(*args, **kwargs) - @_populate_first - def rename(self, old, new): - old_path, old_fn = os.path.split(old) - old_col = self.find(path) - if old_col is None: - raise IOError((errno.ENOENT, "File not found")) - if not isinstance(old_p, Collection): - raise IOError((errno.ENOTDIR, "Parent in path is a file, not a directory")) - if old_fn in old_col: - new_path, new_fn = os.path.split(new) - new_col = self.find(new_path, create=True, create_collection=True) - if not isinstance(new_col, Collection): - raise IOError((errno.ENOTDIR, "Destination is a file, not a directory")) - ent = old_col[old_fn] - del old_col[old_fn] - ent.parent = new_col - new_col[new_fn] = ent - else: - raise IOError((errno.ENOENT, "File not found")) -def import_manifest(manifest_text, into_collection=None, api_client=None, keep=None, num_retries=None): +class Subcollection(SynchronizedCollectionBase): + """This is a subdirectory within a collection that doesn't have its own API + server record. It falls under the umbrella of the root collection.""" + + def __init__(self, parent): + super(Subcollection, self).__init__(parent) + self.lock = self.root_collection().lock + + def root_collection(self): + return self.parent.root_collection() + + def sync_mode(self): + return self.root_collection().sync_mode() + + def _my_api(self): + return self.root_collection()._my_api() + + def _my_keep(self): + return self.root_collection()._my_keep() + + def _my_block_manager(self): + return self.root_collection()._my_block_manager() + + def _populate(self): + self.root_collection()._populate() + + def notify(self, event, collection, name, item): + return self.root_collection().notify(event, collection, name, item) + + @synchronized + def clone(self, new_parent): + c = Subcollection(new_parent) + self._cloneinto(c) + return c + +def import_manifest(manifest_text, + into_collection=None, + api_client=None, + keep=None, + num_retries=None, + sync=SYNC_READONLY): + """Import a manifest into a `Collection`. + + :manifest_text: + The manifest text to import from. + + :into_collection: + The `Collection` that will be initialized (must be empty). + If None, create a new `Collection` object. + + :api_client: + The API client object that will be used when creating a new `Collection` object. + + :keep: + The keep client object that will be used when creating a new `Collection` object. + + :num_retries: + the default number of api client and keep retries on error. + + :sync: + Collection sync mode (only if into_collection is None) + """ if into_collection is not None: if len(into_collection) > 0: raise ArgumentError("Can only import manifest into an empty collection") c = into_collection else: - c = Collection(api_client=api_client, keep_client=keep, num_retries=num_retries) + c = CollectionRoot(api_client=api_client, keep_client=keep, num_retries=num_retries, sync=sync) + + save_sync = c.sync_mode() + c._sync = None STREAM_NAME = 0 BLOCKS = 1 @@ -1010,17 +1555,30 @@ def import_manifest(manifest_text, into_collection=None, api_client=None, keep=N state = STREAM_NAME c.set_unmodified() + c._sync = save_sync return c def export_manifest(item, stream_name=".", portable_locators=False): + """ + :item: + Create a manifest for `item` (must be a `Collection` or `ArvadosFile`). If + `item` is a is a `Collection`, this will also export subcollections. + + :stream_name: + the name of the stream when exporting `item`. + + :portable_locators: + If True, strip any permission hints on block locators. + If False, use block locators as-is. + """ buf = "" - if isinstance(item, Collection): + if isinstance(item, SynchronizedCollectionBase): stream = {} sorted_keys = sorted(item.keys()) for k in [s for s in sorted_keys if isinstance(item[s], ArvadosFile)]: v = item[k] st = [] - for s in v.segments: + for s in v.segments(): loc = s.locator if loc.startswith("bufferblock"): loc = v.parent._my_block_manager()._bufferblocks[loc].locator() @@ -1032,7 +1590,7 @@ def export_manifest(item, stream_name=".", portable_locators=False): if stream: buf += ' '.join(normalize_stream(stream_name, stream)) buf += "\n" - for k in [s for s in sorted_keys if isinstance(item[s], Collection)]: + for k in [s for s in sorted_keys if isinstance(item[s], SynchronizedCollectionBase)]: buf += export_manifest(item[k], stream_name=os.path.join(stream_name, k), portable_locators=portable_locators) elif isinstance(item, ArvadosFile): st = []