X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/991d037e7b4741450dff745d22b0758170fe4ed9..c02ceff00fce94ec5794b53fe890f681acf31121:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index 1c68c8ed43..4be098d351 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -1,4 +1,9 @@ +# Copyright (C) The Arvados Authors. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + from __future__ import absolute_import +from future.utils import listitems, listvalues, viewkeys from builtins import str from past.builtins import basestring from builtins import object @@ -359,7 +364,7 @@ class CollectionWriter(CollectionBase): def portable_data_hash(self): stripped = self.stripped_manifest().encode() - return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped)) + return '{}+{}'.format(hashlib.md5(stripped).hexdigest(), len(stripped)) def manifest_text(self): self.finish_current_stream() @@ -381,6 +386,14 @@ class CollectionWriter(CollectionBase): ret += locators return ret + def save_new(self, name=None): + return self._api_client.collections().create( + ensure_unique_name=True, + body={ + 'name': name, + 'manifest_text': self.manifest_text(), + }).execute(num_retries=self.num_retries) + class ResumableCollectionWriter(CollectionWriter): STATE_PROPS = ['_current_stream_files', '_current_stream_length', @@ -427,7 +440,7 @@ class ResumableCollectionWriter(CollectionWriter): return writer def check_dependencies(self): - for path, orig_stat in list(self._dependencies.items()): + for path, orig_stat in listitems(self._dependencies): if not S_ISREG(orig_stat[ST_MODE]): raise errors.StaleWriterStateError("{} not file".format(path)) try: @@ -621,7 +634,12 @@ class RichCollectionBase(CollectionBase): :path: path to a file in the collection :mode: - one of "r", "r+", "w", "w+", "a", "a+" + a string consisting of "r", "w", or "a", optionally followed + by "b" or "t", optionally followed by "+". + :"b": + binary mode: write() accepts bytes, read() returns bytes. + :"t": + text mode (default): write() accepts strings, read() returns strings. :"r": opens for reading :"r+": @@ -633,33 +651,28 @@ class RichCollectionBase(CollectionBase): 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 errors.ArgumentError("Bad mode '%s'" % mode) - create = (mode != "r") - if create and not self.writable(): - raise IOError(errno.EROFS, "Collection is read only") + if not re.search(r'^[rwa][bt]?\+?$', mode): + raise errors.ArgumentError("Invalid mode {!r}".format(mode)) - if create: - arvfile = self.find_or_create(path, FILE) - else: + if mode[0] == 'r' and '+' not in mode: + fclass = ArvadosFileReader arvfile = self.find(path) + elif not self.writable(): + raise IOError(errno.EROFS, "Collection is read only") + else: + fclass = ArvadosFileWriter + arvfile = self.find_or_create(path, FILE) if arvfile is None: raise IOError(errno.ENOENT, "File not found", path) if not isinstance(arvfile, ArvadosFile): raise IOError(errno.EISDIR, "Is a directory", path) - if mode[0] == "w": + if mode[0] == 'w': arvfile.truncate(0) - name = os.path.basename(path) - - if mode == "r": - return ArvadosFileReader(arvfile, num_retries=self.num_retries) - else: - return ArvadosFileWriter(arvfile, mode, num_retries=self.num_retries) + return fclass(arvfile, mode=mode, num_retries=self.num_retries) def modified(self): """Determine if the collection has been modified since last commited.""" @@ -681,7 +694,7 @@ class RichCollectionBase(CollectionBase): if value == self._committed: return if value: - for k,v in list(self._items.items()): + for k,v in listitems(self._items): v.set_committed(True) self._committed = True else: @@ -692,7 +705,7 @@ class RichCollectionBase(CollectionBase): @synchronized def __iter__(self): """Iterate over names of files and collections contained in this collection.""" - return iter(list(self._items.keys())) + return iter(viewkeys(self._items)) @synchronized def __getitem__(self, k): @@ -724,17 +737,17 @@ class RichCollectionBase(CollectionBase): @synchronized def keys(self): """Get a list of names of files and collections directly contained in this collection.""" - return list(self._items.keys()) + return self._items.keys() @synchronized def values(self): """Get a list of files and collection objects directly contained in this collection.""" - return list(self._items.values()) + return listvalues(self._items) @synchronized def items(self): """Get a list of (name, object) tuples directly contained in this collection.""" - return list(self._items.items()) + return listitems(self._items) def exists(self, path): """Test if there is a file or collection at `path`.""" @@ -767,7 +780,7 @@ class RichCollectionBase(CollectionBase): item.remove(pathcomponents[1]) def _clonefrom(self, source): - for k,v in list(source.items()): + for k,v in listitems(source): self._items[k] = v.clone(self, k) def clone(self): @@ -1084,7 +1097,7 @@ class RichCollectionBase(CollectionBase): return self._portable_data_hash else: stripped = self.portable_manifest_text().encode() - return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped)) + return '{}+{}'.format(hashlib.md5(stripped).hexdigest(), len(stripped)) @synchronized def subscribe(self, callback): @@ -1125,7 +1138,7 @@ class RichCollectionBase(CollectionBase): @synchronized def flush(self): """Flush bufferblocks to Keep.""" - for e in list(self.values()): + for e in listvalues(self): e.flush() @@ -1180,8 +1193,9 @@ class Collection(RichCollectionBase): """Collection constructor. :manifest_locator_or_text: - One of Arvados collection UUID, block locator of - a manifest, raw manifest text, or None (to create an empty collection). + An Arvados collection UUID, portable data hash, raw manifest + text, or (if creating an empty collection) None. + :parent: the parent Collection, may be None. @@ -1320,65 +1334,25 @@ class Collection(RichCollectionBase): # it. If instantiation fails, we'll fall back to the except # clause, just like any other Collection lookup # failure. Return an exception, or None if successful. - try: - self._remember_api_response(self._my_api().collections().get( - uuid=self._manifest_locator).execute( - num_retries=self.num_retries)) - self._manifest_text = self._api_response['manifest_text'] - self._portable_data_hash = self._api_response['portable_data_hash'] - # If not overriden via kwargs, we should try to load the - # replication_desired from the API server - if self.replication_desired is None: - self.replication_desired = self._api_response.get('replication_desired', None) - return None - except Exception as e: - return e - - def _populate_from_keep(self): - # Retrieve a manifest directly from Keep. This has a chance of - # working if [a] the locator includes a permission signature - # or [b] the Keep services are operating in world-readable - # mode. Return an exception, or None if successful. - try: - self._manifest_text = self._my_keep().get( - self._manifest_locator, num_retries=self.num_retries).decode() - except Exception as e: - return e + self._remember_api_response(self._my_api().collections().get( + uuid=self._manifest_locator).execute( + num_retries=self.num_retries)) + self._manifest_text = self._api_response['manifest_text'] + self._portable_data_hash = self._api_response['portable_data_hash'] + # If not overriden via kwargs, we should try to load the + # replication_desired from the API server + if self.replication_desired is None: + self.replication_desired = self._api_response.get('replication_desired', None) def _populate(self): - 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 - arvados.util.keep_locator_pattern.match( - self._manifest_locator)) - if ((self._manifest_text is None) and - arvados.util.signed_locator_pattern.match(self._manifest_locator)): - error_via_keep = self._populate_from_keep() if self._manifest_text is None: - error_via_api = self._populate_from_api_server() - if error_via_api is not None and not should_try_keep: - raise error_via_api - if ((self._manifest_text is None) and - not error_via_keep and - should_try_keep): - # Looks like a keep locator, and we didn't already try keep above - error_via_keep = self._populate_from_keep() - if self._manifest_text is None: - # Nothing worked! - raise errors.NotFoundError( - ("Failed to retrieve collection '{}' " + - "from either API server ({}) or Keep ({})." - ).format( - self._manifest_locator, - error_via_api, - error_via_keep)) - # populate + if self._manifest_locator is None: + return + else: + self._populate_from_api_server() self._baseline_manifest = self._manifest_text self._import_manifest(self._manifest_text) - def _has_collection_uuid(self): return self._manifest_locator is not None and re.match(arvados.util.collection_uuid_pattern, self._manifest_locator) @@ -1679,9 +1653,8 @@ class Subcollection(RichCollectionBase): class CollectionReader(Collection): """A read-only collection object. - Initialize from an api collection record locator, a portable data hash of a - manifest, or raw manifest text. See `Collection` constructor for detailed - options. + Initialize from a collection UUID or portable data hash, or raw + manifest text. See `Collection` constructor for detailed options. """ def __init__(self, manifest_locator_or_text, *args, **kwargs):