10 from collections import deque
13 from .arvfile import split, _FileLikeObjectBase, ArvadosFile, ArvadosFileWriter, ArvadosFileReader, _BlockManager, synchronized, must_be_writable, NoopLock
14 from keep import KeepLocator, KeepClient
15 from .stream import StreamReader
16 from ._normalize_stream import normalize_stream
17 from ._ranges import Range, LocatorAndRange
18 from .safeapi import ThreadSafeApiCache
23 from arvados.retry import retry_method
25 _logger = logging.getLogger('arvados.collection')
27 class CollectionBase(object):
31 def __exit__(self, exc_type, exc_value, traceback):
35 if self._keep_client is None:
36 self._keep_client = KeepClient(api_client=self._api_client,
37 num_retries=self.num_retries)
38 return self._keep_client
40 def stripped_manifest(self):
41 """Get the manifest with locator hints stripped.
43 Return the manifest for the current collection with all
44 non-portable hints (i.e., permission signatures and other
45 hints other than size hints) removed from the locators.
47 raw = self.manifest_text()
49 for line in raw.split("\n"):
52 clean_fields = fields[:1] + [
53 (re.sub(r'\+[^\d][^\+]*', '', x)
54 if re.match(util.keep_locator_pattern, x)
57 clean += [' '.join(clean_fields), "\n"]
61 class _WriterFile(_FileLikeObjectBase):
62 def __init__(self, coll_writer, name):
63 super(_WriterFile, self).__init__(name, 'wb')
64 self.dest = coll_writer
67 super(_WriterFile, self).close()
68 self.dest.finish_current_file()
70 @_FileLikeObjectBase._before_close
71 def write(self, data):
74 @_FileLikeObjectBase._before_close
75 def writelines(self, seq):
79 @_FileLikeObjectBase._before_close
81 self.dest.flush_data()
84 class CollectionWriter(CollectionBase):
85 def __init__(self, api_client=None, num_retries=0, replication=None):
86 """Instantiate a CollectionWriter.
88 CollectionWriter lets you build a new Arvados Collection from scratch.
89 Write files to it. The CollectionWriter will upload data to Keep as
90 appropriate, and provide you with the Collection manifest text when
94 * api_client: The API client to use to look up Collections. If not
95 provided, CollectionReader will build one from available Arvados
97 * num_retries: The default number of times to retry failed
98 service requests. Default 0. You may change this value
99 after instantiation, but note those changes may not
100 propagate to related objects like the Keep client.
101 * replication: The number of copies of each block to store.
102 If this argument is None or not supplied, replication is
103 the server-provided default if available, otherwise 2.
105 self._api_client = api_client
106 self.num_retries = num_retries
107 self.replication = (2 if replication is None else replication)
108 self._keep_client = None
109 self._data_buffer = []
110 self._data_buffer_len = 0
111 self._current_stream_files = []
112 self._current_stream_length = 0
113 self._current_stream_locators = []
114 self._current_stream_name = '.'
115 self._current_file_name = None
116 self._current_file_pos = 0
117 self._finished_streams = []
118 self._close_file = None
119 self._queued_file = None
120 self._queued_dirents = deque()
121 self._queued_trees = deque()
122 self._last_open = None
124 def __exit__(self, exc_type, exc_value, traceback):
128 def do_queued_work(self):
129 # The work queue consists of three pieces:
130 # * _queued_file: The file object we're currently writing to the
132 # * _queued_dirents: Entries under the current directory
133 # (_queued_trees[0]) that we want to write or recurse through.
134 # This may contain files from subdirectories if
135 # max_manifest_depth == 0 for this directory.
136 # * _queued_trees: Directories that should be written as separate
137 # streams to the Collection.
138 # This function handles the smallest piece of work currently queued
139 # (current file, then current directory, then next directory) until
140 # no work remains. The _work_THING methods each do a unit of work on
141 # THING. _queue_THING methods add a THING to the work queue.
143 if self._queued_file:
145 elif self._queued_dirents:
147 elif self._queued_trees:
152 def _work_file(self):
154 buf = self._queued_file.read(config.KEEP_BLOCK_SIZE)
158 self.finish_current_file()
160 self._queued_file.close()
161 self._close_file = None
162 self._queued_file = None
164 def _work_dirents(self):
165 path, stream_name, max_manifest_depth = self._queued_trees[0]
166 if stream_name != self.current_stream_name():
167 self.start_new_stream(stream_name)
168 while self._queued_dirents:
169 dirent = self._queued_dirents.popleft()
170 target = os.path.join(path, dirent)
171 if os.path.isdir(target):
172 self._queue_tree(target,
173 os.path.join(stream_name, dirent),
174 max_manifest_depth - 1)
176 self._queue_file(target, dirent)
178 if not self._queued_dirents:
179 self._queued_trees.popleft()
181 def _work_trees(self):
182 path, stream_name, max_manifest_depth = self._queued_trees[0]
183 d = util.listdir_recursive(
184 path, max_depth = (None if max_manifest_depth == 0 else 0))
186 self._queue_dirents(stream_name, d)
188 self._queued_trees.popleft()
190 def _queue_file(self, source, filename=None):
191 assert (self._queued_file is None), "tried to queue more than one file"
192 if not hasattr(source, 'read'):
193 source = open(source, 'rb')
194 self._close_file = True
196 self._close_file = False
198 filename = os.path.basename(source.name)
199 self.start_new_file(filename)
200 self._queued_file = source
202 def _queue_dirents(self, stream_name, dirents):
203 assert (not self._queued_dirents), "tried to queue more than one tree"
204 self._queued_dirents = deque(sorted(dirents))
206 def _queue_tree(self, path, stream_name, max_manifest_depth):
207 self._queued_trees.append((path, stream_name, max_manifest_depth))
209 def write_file(self, source, filename=None):
210 self._queue_file(source, filename)
211 self.do_queued_work()
213 def write_directory_tree(self,
214 path, stream_name='.', max_manifest_depth=-1):
215 self._queue_tree(path, stream_name, max_manifest_depth)
216 self.do_queued_work()
218 def write(self, newdata):
219 if hasattr(newdata, '__iter__'):
223 self._data_buffer.append(newdata)
224 self._data_buffer_len += len(newdata)
225 self._current_stream_length += len(newdata)
226 while self._data_buffer_len >= config.KEEP_BLOCK_SIZE:
229 def open(self, streampath, filename=None):
230 """open(streampath[, filename]) -> file-like object
232 Pass in the path of a file to write to the Collection, either as a
233 single string or as two separate stream name and file name arguments.
234 This method returns a file-like object you can write to add it to the
237 You may only have one file object from the Collection open at a time,
238 so be sure to close the object when you're done. Using the object in
239 a with statement makes that easy::
241 with cwriter.open('./doc/page1.txt') as outfile:
242 outfile.write(page1_data)
243 with cwriter.open('./doc/page2.txt') as outfile:
244 outfile.write(page2_data)
247 streampath, filename = split(streampath)
248 if self._last_open and not self._last_open.closed:
249 raise errors.AssertionError(
250 "can't open '{}' when '{}' is still open".format(
251 filename, self._last_open.name))
252 if streampath != self.current_stream_name():
253 self.start_new_stream(streampath)
254 self.set_current_file_name(filename)
255 self._last_open = _WriterFile(self, filename)
256 return self._last_open
258 def flush_data(self):
259 data_buffer = ''.join(self._data_buffer)
261 self._current_stream_locators.append(
263 data_buffer[0:config.KEEP_BLOCK_SIZE],
264 copies=self.replication))
265 self._data_buffer = [data_buffer[config.KEEP_BLOCK_SIZE:]]
266 self._data_buffer_len = len(self._data_buffer[0])
268 def start_new_file(self, newfilename=None):
269 self.finish_current_file()
270 self.set_current_file_name(newfilename)
272 def set_current_file_name(self, newfilename):
273 if re.search(r'[\t\n]', newfilename):
274 raise errors.AssertionError(
275 "Manifest filenames cannot contain whitespace: %s" %
277 elif re.search(r'\x00', newfilename):
278 raise errors.AssertionError(
279 "Manifest filenames cannot contain NUL characters: %s" %
281 self._current_file_name = newfilename
283 def current_file_name(self):
284 return self._current_file_name
286 def finish_current_file(self):
287 if self._current_file_name is None:
288 if self._current_file_pos == self._current_stream_length:
290 raise errors.AssertionError(
291 "Cannot finish an unnamed file " +
292 "(%d bytes at offset %d in '%s' stream)" %
293 (self._current_stream_length - self._current_file_pos,
294 self._current_file_pos,
295 self._current_stream_name))
296 self._current_stream_files.append([
297 self._current_file_pos,
298 self._current_stream_length - self._current_file_pos,
299 self._current_file_name])
300 self._current_file_pos = self._current_stream_length
301 self._current_file_name = None
303 def start_new_stream(self, newstreamname='.'):
304 self.finish_current_stream()
305 self.set_current_stream_name(newstreamname)
307 def set_current_stream_name(self, newstreamname):
308 if re.search(r'[\t\n]', newstreamname):
309 raise errors.AssertionError(
310 "Manifest stream names cannot contain whitespace")
311 self._current_stream_name = '.' if newstreamname=='' else newstreamname
313 def current_stream_name(self):
314 return self._current_stream_name
316 def finish_current_stream(self):
317 self.finish_current_file()
319 if not self._current_stream_files:
321 elif self._current_stream_name is None:
322 raise errors.AssertionError(
323 "Cannot finish an unnamed stream (%d bytes in %d files)" %
324 (self._current_stream_length, len(self._current_stream_files)))
326 if not self._current_stream_locators:
327 self._current_stream_locators.append(config.EMPTY_BLOCK_LOCATOR)
328 self._finished_streams.append([self._current_stream_name,
329 self._current_stream_locators,
330 self._current_stream_files])
331 self._current_stream_files = []
332 self._current_stream_length = 0
333 self._current_stream_locators = []
334 self._current_stream_name = None
335 self._current_file_pos = 0
336 self._current_file_name = None
339 """Store the manifest in Keep and return its locator.
341 This is useful for storing manifest fragments (task outputs)
342 temporarily in Keep during a Crunch job.
344 In other cases you should make a collection instead, by
345 sending manifest_text() to the API server's "create
346 collection" endpoint.
348 return self._my_keep().put(self.manifest_text(), copies=self.replication)
350 def portable_data_hash(self):
351 stripped = self.stripped_manifest()
352 return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped))
354 def manifest_text(self):
355 self.finish_current_stream()
358 for stream in self._finished_streams:
359 if not re.search(r'^\.(/.*)?$', stream[0]):
361 manifest += stream[0].replace(' ', '\\040')
362 manifest += ' ' + ' '.join(stream[1])
363 manifest += ' ' + ' '.join("%d:%d:%s" % (sfile[0], sfile[1], sfile[2].replace(' ', '\\040')) for sfile in stream[2])
368 def data_locators(self):
370 for name, locators, files in self._finished_streams:
375 class ResumableCollectionWriter(CollectionWriter):
376 STATE_PROPS = ['_current_stream_files', '_current_stream_length',
377 '_current_stream_locators', '_current_stream_name',
378 '_current_file_name', '_current_file_pos', '_close_file',
379 '_data_buffer', '_dependencies', '_finished_streams',
380 '_queued_dirents', '_queued_trees']
382 def __init__(self, api_client=None, **kwargs):
383 self._dependencies = {}
384 super(ResumableCollectionWriter, self).__init__(api_client, **kwargs)
387 def from_state(cls, state, *init_args, **init_kwargs):
388 # Try to build a new writer from scratch with the given state.
389 # If the state is not suitable to resume (because files have changed,
390 # been deleted, aren't predictable, etc.), raise a
391 # StaleWriterStateError. Otherwise, return the initialized writer.
392 # The caller is responsible for calling writer.do_queued_work()
393 # appropriately after it's returned.
394 writer = cls(*init_args, **init_kwargs)
395 for attr_name in cls.STATE_PROPS:
396 attr_value = state[attr_name]
397 attr_class = getattr(writer, attr_name).__class__
398 # Coerce the value into the same type as the initial value, if
400 if attr_class not in (type(None), attr_value.__class__):
401 attr_value = attr_class(attr_value)
402 setattr(writer, attr_name, attr_value)
403 # Check dependencies before we try to resume anything.
404 if any(KeepLocator(ls).permission_expired()
405 for ls in writer._current_stream_locators):
406 raise errors.StaleWriterStateError(
407 "locators include expired permission hint")
408 writer.check_dependencies()
409 if state['_current_file'] is not None:
410 path, pos = state['_current_file']
412 writer._queued_file = open(path, 'rb')
413 writer._queued_file.seek(pos)
414 except IOError as error:
415 raise errors.StaleWriterStateError(
416 "failed to reopen active file {}: {}".format(path, error))
419 def check_dependencies(self):
420 for path, orig_stat in self._dependencies.items():
421 if not S_ISREG(orig_stat[ST_MODE]):
422 raise errors.StaleWriterStateError("{} not file".format(path))
424 now_stat = tuple(os.stat(path))
425 except OSError as error:
426 raise errors.StaleWriterStateError(
427 "failed to stat {}: {}".format(path, error))
428 if ((not S_ISREG(now_stat[ST_MODE])) or
429 (orig_stat[ST_MTIME] != now_stat[ST_MTIME]) or
430 (orig_stat[ST_SIZE] != now_stat[ST_SIZE])):
431 raise errors.StaleWriterStateError("{} changed".format(path))
433 def dump_state(self, copy_func=lambda x: x):
434 state = {attr: copy_func(getattr(self, attr))
435 for attr in self.STATE_PROPS}
436 if self._queued_file is None:
437 state['_current_file'] = None
439 state['_current_file'] = (os.path.realpath(self._queued_file.name),
440 self._queued_file.tell())
443 def _queue_file(self, source, filename=None):
445 src_path = os.path.realpath(source)
447 raise errors.AssertionError("{} not a file path".format(source))
449 path_stat = os.stat(src_path)
450 except OSError as stat_error:
452 super(ResumableCollectionWriter, self)._queue_file(source, filename)
453 fd_stat = os.fstat(self._queued_file.fileno())
454 if not S_ISREG(fd_stat.st_mode):
455 # We won't be able to resume from this cache anyway, so don't
456 # worry about further checks.
457 self._dependencies[source] = tuple(fd_stat)
458 elif path_stat is None:
459 raise errors.AssertionError(
460 "could not stat {}: {}".format(source, stat_error))
461 elif path_stat.st_ino != fd_stat.st_ino:
462 raise errors.AssertionError(
463 "{} changed between open and stat calls".format(source))
465 self._dependencies[src_path] = tuple(fd_stat)
467 def write(self, data):
468 if self._queued_file is None:
469 raise errors.AssertionError(
470 "resumable writer can't accept unsourced data")
471 return super(ResumableCollectionWriter, self).write(data)
478 COLLECTION = "collection"
480 class RichCollectionBase(CollectionBase):
481 """Base class for Collections and Subcollections.
483 Implements the majority of functionality relating to accessing items in the
488 def __init__(self, parent=None):
490 self._modified = True
491 self._callback = None
495 raise NotImplementedError()
498 raise NotImplementedError()
500 def _my_block_manager(self):
501 raise NotImplementedError()
504 raise NotImplementedError()
506 def root_collection(self):
507 raise NotImplementedError()
509 def notify(self, event, collection, name, item):
510 raise NotImplementedError()
512 def stream_name(self):
513 raise NotImplementedError()
517 def find_or_create(self, path, create_type):
518 """Recursively search the specified file path.
520 May return either a `Collection` or `ArvadosFile`. If not found, will
521 create a new item at the specified path based on `create_type`. Will
522 create intermediate subcollections needed to contain the final item in
526 One of `arvados.collection.FILE` or
527 `arvados.collection.COLLECTION`. If the path is not found, and value
528 of create_type is FILE then create and return a new ArvadosFile for
529 the last path component. If COLLECTION, then create and return a new
530 Collection for the last path component.
534 pathcomponents = path.split("/", 1)
535 if pathcomponents[0]:
536 item = self._items.get(pathcomponents[0])
537 if len(pathcomponents) == 1:
540 if create_type == COLLECTION:
541 item = Subcollection(self, pathcomponents[0])
543 item = ArvadosFile(self, pathcomponents[0])
544 self._items[pathcomponents[0]] = item
545 self._modified = True
546 self.notify(ADD, self, pathcomponents[0], item)
550 # create new collection
551 item = Subcollection(self, pathcomponents[0])
552 self._items[pathcomponents[0]] = item
553 self._modified = True
554 self.notify(ADD, self, pathcomponents[0], item)
555 if isinstance(item, RichCollectionBase):
556 return item.find_or_create(pathcomponents[1], create_type)
558 raise IOError(errno.ENOTDIR, "Interior path components must be subcollection")
563 def find(self, path):
564 """Recursively search the specified file path.
566 May return either a Collection or ArvadosFile. Return None if not
571 raise errors.ArgumentError("Parameter 'path' must not be empty.")
573 pathcomponents = path.split("/", 1)
574 item = self._items.get(pathcomponents[0])
575 if len(pathcomponents) == 1:
578 if isinstance(item, RichCollectionBase):
579 if pathcomponents[1]:
580 return item.find(pathcomponents[1])
584 raise IOError(errno.ENOTDIR, "Interior path components must be subcollection")
586 def mkdirs(self, path):
587 """Recursive subcollection create.
589 Like `os.mkdirs()`. Will create intermediate subcollections needed to
590 contain the leaf subcollection path.
593 return self.find_or_create(path, COLLECTION)
595 def open(self, path, mode="r"):
596 """Open a file-like object for access.
599 path to a file in the collection
601 one of "r", "r+", "w", "w+", "a", "a+"
605 opens for reading and writing. Reads/writes share a file pointer.
607 truncates to 0 and opens for reading and writing. Reads/writes share a file pointer.
609 opens for reading and writing. All writes are appended to
610 the end of the file. Writing does not affect the file pointer for
613 mode = mode.replace("b", "")
614 if len(mode) == 0 or mode[0] not in ("r", "w", "a"):
615 raise errors.ArgumentError("Bad mode '%s'" % mode)
616 create = (mode != "r")
618 if create and not self.writable():
619 raise IOError(errno.EROFS, "Collection is read only")
622 arvfile = self.find_or_create(path, FILE)
624 arvfile = self.find(path)
627 raise IOError(errno.ENOENT, "File not found")
628 if not isinstance(arvfile, ArvadosFile):
629 raise IOError(errno.EISDIR, "Path must refer to a file.")
634 name = os.path.basename(path)
637 return ArvadosFileReader(arvfile, mode, num_retries=self.num_retries)
639 return ArvadosFileWriter(arvfile, mode, num_retries=self.num_retries)
643 """Test if the collection (or any subcollection or file) has been modified."""
646 for k,v in self._items.items():
652 def set_unmodified(self):
653 """Recursively clear modified flag."""
654 self._modified = False
655 for k,v in self._items.items():
660 """Iterate over names of files and collections contained in this collection."""
661 return iter(self._items.keys())
664 def __getitem__(self, k):
665 """Get a file or collection that is directly contained by this collection.
667 If you want to search a path, use `find()` instead.
670 return self._items[k]
673 def __contains__(self, k):
674 """Test if there is a file or collection a directly contained by this collection."""
675 return k in self._items
679 """Get the number of items directly contained in this collection."""
680 return len(self._items)
684 def __delitem__(self, p):
685 """Delete an item by name which is directly contained by this collection."""
687 self._modified = True
688 self.notify(DEL, self, p, None)
692 """Get a list of names of files and collections directly contained in this collection."""
693 return self._items.keys()
697 """Get a list of files and collection objects directly contained in this collection."""
698 return self._items.values()
702 """Get a list of (name, object) tuples directly contained in this collection."""
703 return self._items.items()
705 def exists(self, path):
706 """Test if there is a file or collection at `path`."""
707 return self.find(path) is not None
711 def remove(self, path, recursive=False):
712 """Remove the file or subcollection (directory) at `path`.
715 Specify whether to remove non-empty subcollections (True), or raise an error (False).
719 raise errors.ArgumentError("Parameter 'path' must not be empty.")
721 pathcomponents = path.split("/", 1)
722 item = self._items.get(pathcomponents[0])
724 raise IOError(errno.ENOENT, "File not found")
725 if len(pathcomponents) == 1:
726 if isinstance(self._items[pathcomponents[0]], RichCollectionBase) and len(self._items[pathcomponents[0]]) > 0 and not recursive:
727 raise IOError(errno.ENOTEMPTY, "Subcollection not empty")
728 deleteditem = self._items[pathcomponents[0]]
729 del self._items[pathcomponents[0]]
730 self._modified = True
731 self.notify(DEL, self, pathcomponents[0], deleteditem)
733 item.remove(pathcomponents[1])
735 def _clonefrom(self, source):
736 for k,v in source.items():
737 self._items[k] = v.clone(self, k)
740 raise NotImplementedError()
744 def add(self, source_obj, target_name, overwrite=False, reparent=False):
745 """Copy or move a file or subcollection to this collection.
748 An ArvadosFile, or Subcollection object
751 Destination item name. If the target name already exists and is a
752 file, this will raise an error unless you specify `overwrite=True`.
755 Whether to overwrite target file if it already exists.
758 If True, source_obj will be moved from its parent collection to this collection.
759 If False, source_obj will be copied and the parent collection will be
764 if target_name in self and not overwrite:
765 raise IOError(errno.EEXIST, "File already exists")
768 if target_name in self:
769 modified_from = self[target_name]
771 # Actually make the move or copy.
773 source_obj.reparent(self, target_name)
776 item = source_obj.clone(self, target_name)
778 self._items[target_name] = item
779 self._modified = True
782 self.notify(MOD, self, target_name, (modified_from, item))
784 self.notify(ADD, self, target_name, item)
786 def _get_src_target(self, source, target_path, source_collection, create_dest):
787 if source_collection is None:
788 source_collection = self
791 if isinstance(source, basestring):
792 source_obj = source_collection.find(source)
793 if source_obj is None:
794 raise IOError(errno.ENOENT, "File not found")
795 sourcecomponents = source.split("/")
798 sourcecomponents = None
800 # Find parent collection the target path
801 targetcomponents = target_path.split("/")
803 # Determine the name to use.
804 target_name = targetcomponents[-1] if targetcomponents[-1] else sourcecomponents[-1]
807 raise errors.ArgumentError("Target path is empty and source is an object. Cannot determine destination filename to use.")
810 target_dir = self.find_or_create("/".join(targetcomponents[0:-1]), COLLECTION)
812 if len(targetcomponents) > 1:
813 target_dir = self.find("/".join(targetcomponents[0:-1]))
817 if target_dir is None:
818 raise IOError(errno.ENOENT, "Target directory not found.")
820 if target_name in target_dir and isinstance(self[target_name], RichCollectionBase) and sourcecomponents:
821 target_dir = target_dir[target_name]
822 target_name = sourcecomponents[-1]
824 return (source_obj, target_dir, target_name)
828 def copy(self, source, target_path, source_collection=None, overwrite=False):
829 """Copy a file or subcollection to a new path in this collection.
832 A string with a path to source file or subcollection, or an actual ArvadosFile or Subcollection object.
835 Destination file or path. If the target path already exists and is a
836 subcollection, the item will be placed inside the subcollection. If
837 the target path already exists and is a file, this will raise an error
838 unless you specify `overwrite=True`.
841 Collection to copy `source_path` from (default `self`)
844 Whether to overwrite target file if it already exists.
847 source_obj, target_dir, target_name = self._get_src_target(source, target_path, source_collection, True)
848 target_dir.add(source_obj, target_name, overwrite, False)
852 def rename(self, source, target_path, source_collection=None, overwrite=False):
853 """Move a file or subcollection from `source_collection` to a new path in this collection.
856 A string with a path to source file or subcollection.
859 Destination file or path. If the target path already exists and is a
860 subcollection, the item will be placed inside the subcollection. If
861 the target path already exists and is a file, this will raise an error
862 unless you specify `overwrite=True`.
865 Collection to copy `source_path` from (default `self`)
868 Whether to overwrite target file if it already exists.
871 source_obj, target_dir, target_name = self._get_src_target(source, target_path, source_collection, False)
872 if not source_obj.writable():
873 raise IOError(errno.EROFS, "Source collection must be writable.")
874 target_dir.add(source_obj, target_name, overwrite, True)
876 def portable_manifest_text(self, stream_name="."):
877 """Get the manifest text for this collection, sub collections and files.
879 This method does not flush outstanding blocks to Keep. It will return
880 a normalized manifest with access tokens stripped.
883 Name to use for this stream (directory)
886 return self._get_manifest_text(stream_name, True, True)
888 def manifest_text(self, stream_name=".", strip=False, normalize=False):
889 """Get the manifest text for this collection, sub collections and files.
891 This method will flush outstanding blocks to Keep. By default, it will
892 not normalize an unmodified manifest or strip access tokens.
895 Name to use for this stream (directory)
898 If True, remove signing tokens from block locators if present.
899 If False (default), block locators are left unchanged.
902 If True, always export the manifest text in normalized form
903 even if the Collection is not modified. If False (default) and the collection
904 is not modified, return the original manifest text even if it is not
909 self._my_block_manager().commit_all()
910 return self._get_manifest_text(stream_name, strip, normalize)
913 def _get_manifest_text(self, stream_name, strip, normalize):
914 """Get the manifest text for this collection, sub collections and files.
917 Name to use for this stream (directory)
920 If True, remove signing tokens from block locators if present.
921 If False (default), block locators are left unchanged.
924 If True, always export the manifest text in normalized form
925 even if the Collection is not modified. If False (default) and the collection
926 is not modified, return the original manifest text even if it is not
931 if self.modified() or self._manifest_text is None or normalize:
934 sorted_keys = sorted(self.keys())
935 for filename in [s for s in sorted_keys if isinstance(self[s], ArvadosFile)]:
936 # Create a stream per file `k`
937 arvfile = self[filename]
939 for segment in arvfile.segments():
940 loc = segment.locator
941 if arvfile.parent._my_block_manager().is_bufferblock(loc):
942 loc = arvfile.parent._my_block_manager().get_bufferblock(loc).locator()
944 loc = KeepLocator(loc).stripped()
945 filestream.append(LocatorAndRange(loc, KeepLocator(loc).size,
946 segment.segment_offset, segment.range_size))
947 stream[filename] = filestream
949 buf.append(" ".join(normalize_stream(stream_name, stream)) + "\n")
950 for dirname in [s for s in sorted_keys if isinstance(self[s], RichCollectionBase)]:
951 buf.append(self[dirname].manifest_text(stream_name=os.path.join(stream_name, dirname), strip=strip, normalize=True))
955 return self.stripped_manifest()
957 return self._manifest_text
960 def diff(self, end_collection, prefix=".", holding_collection=None):
961 """Generate list of add/modify/delete actions.
963 When given to `apply`, will change `self` to match `end_collection`
967 if holding_collection is None:
968 holding_collection = Collection(api_client=self._my_api(), keep_client=self._my_keep())
970 if k not in end_collection:
971 changes.append((DEL, os.path.join(prefix, k), self[k].clone(holding_collection, "")))
972 for k in end_collection:
974 if isinstance(end_collection[k], Subcollection) and isinstance(self[k], Subcollection):
975 changes.extend(self[k].diff(end_collection[k], os.path.join(prefix, k), holding_collection))
976 elif end_collection[k] != self[k]:
977 changes.append((MOD, os.path.join(prefix, k), self[k].clone(holding_collection, ""), end_collection[k].clone(holding_collection, "")))
979 changes.append((ADD, os.path.join(prefix, k), end_collection[k].clone(holding_collection, "")))
984 def apply(self, changes):
985 """Apply changes from `diff`.
987 If a change conflicts with a local change, it will be saved to an
988 alternate path indicating the conflict.
991 for change in changes:
992 event_type = change[0]
995 local = self.find(path)
996 conflictpath = "%s~conflict-%s~" % (path, time.strftime("%Y-%m-%d-%H:%M:%S",
998 if event_type == ADD:
1000 # No local file at path, safe to copy over new file
1001 self.copy(initial, path)
1002 elif local is not None and local != initial:
1003 # There is already local file and it is different:
1004 # save change to conflict file.
1005 self.copy(initial, conflictpath)
1006 elif event_type == MOD:
1008 if local == initial:
1009 # Local matches the "initial" item so it has not
1010 # changed locally and is safe to update.
1011 if isinstance(local, ArvadosFile) and isinstance(final, ArvadosFile):
1012 # Replace contents of local file with new contents
1013 local.replace_contents(final)
1015 # Overwrite path with new item; this can happen if
1016 # path was a file and is now a collection or vice versa
1017 self.copy(final, path, overwrite=True)
1019 # Local is missing (presumably deleted) or local doesn't
1020 # match the "start" value, so save change to conflict file
1021 self.copy(final, conflictpath)
1022 elif event_type == DEL:
1023 if local == initial:
1024 # Local item matches "initial" value, so it is safe to remove.
1025 self.remove(path, recursive=True)
1026 # else, the file is modified or already removed, in either
1027 # case we don't want to try to remove it.
1029 def portable_data_hash(self):
1030 """Get the portable data hash for this collection's manifest."""
1031 stripped = self.portable_manifest_text()
1032 return hashlib.md5(stripped).hexdigest() + '+' + str(len(stripped))
1035 def subscribe(self, callback):
1036 if self._callback is None:
1037 self._callback = callback
1039 raise errors.ArgumentError("A callback is already set on this collection.")
1042 def unsubscribe(self):
1043 if self._callback is not None:
1044 self._callback = None
1047 def notify(self, event, collection, name, item):
1049 self._callback(event, collection, name, item)
1050 self.root_collection().notify(event, collection, name, item)
1053 def __eq__(self, other):
1056 if not isinstance(other, RichCollectionBase):
1058 if len(self._items) != len(other):
1060 for k in self._items:
1063 if self._items[k] != other[k]:
1067 def __ne__(self, other):
1068 return not self.__eq__(other)
1071 class Collection(RichCollectionBase):
1072 """Represents the root of an Arvados Collection.
1074 This class is threadsafe. The root collection object, all subcollections
1075 and files are protected by a single lock (i.e. each access locks the entire
1081 :To read an existing file:
1082 `c.open("myfile", "r")`
1084 :To write a new file:
1085 `c.open("myfile", "w")`
1087 :To determine if a file exists:
1088 `c.find("myfile") is not None`
1091 `c.copy("source", "dest")`
1094 `c.remove("myfile")`
1096 :To save to an existing collection record:
1099 :To save a new collection record:
1102 :To merge remote changes into this object:
1105 Must be associated with an API server Collection record (during
1106 initialization, or using `save_new`) to use `save` or `update`
1110 def __init__(self, manifest_locator_or_text=None,
1116 block_manager=None):
1117 """Collection constructor.
1119 :manifest_locator_or_text:
1120 One of Arvados collection UUID, block locator of
1121 a manifest, raw manifest text, or None (to create an empty collection).
1123 the parent Collection, may be None.
1125 A dict containing keys for ARVADOS_API_HOST and ARVADOS_API_TOKEN.
1126 Prefer this over supplying your own api_client and keep_client (except in testing).
1127 Will use default config settings if not specified.
1129 The API client object to use for requests. If not specified, create one using `apiconfig`.
1131 the Keep client to use for requests. If not specified, create one using `apiconfig`.
1133 the number of retries for API and Keep requests.
1135 the block manager to use. If not specified, create one.
1138 super(Collection, self).__init__(parent)
1139 self._api_client = api_client
1140 self._keep_client = keep_client
1141 self._block_manager = block_manager
1144 self._config = apiconfig
1146 self._config = config.settings()
1148 self.num_retries = num_retries if num_retries is not None else 0
1149 self._manifest_locator = None
1150 self._manifest_text = None
1151 self._api_response = None
1153 self.lock = threading.RLock()
1156 if manifest_locator_or_text:
1157 if re.match(util.keep_locator_pattern, manifest_locator_or_text):
1158 self._manifest_locator = manifest_locator_or_text
1159 elif re.match(util.collection_uuid_pattern, manifest_locator_or_text):
1160 self._manifest_locator = manifest_locator_or_text
1161 elif re.match(util.manifest_pattern, manifest_locator_or_text):
1162 self._manifest_text = manifest_locator_or_text
1164 raise errors.ArgumentError(
1165 "Argument to CollectionReader must be a manifest or a collection UUID")
1169 except (IOError, errors.SyntaxError) as e:
1170 raise errors.ArgumentError("Error processing manifest text: %s", e)
1172 def root_collection(self):
1175 def stream_name(self):
1183 def update(self, other=None, num_retries=None):
1184 """Merge the latest collection on the API server with the current collection."""
1187 if self._manifest_locator is None:
1188 raise errors.ArgumentError("`other` is None but collection does not have a manifest_locator uuid")
1189 response = self._my_api().collections().get(uuid=self._manifest_locator).execute(num_retries=num_retries)
1190 other = CollectionReader(response["manifest_text"])
1191 baseline = CollectionReader(self._manifest_text)
1192 self.apply(baseline.diff(other))
1196 if self._api_client is None:
1197 self._api_client = ThreadSafeApiCache(self._config)
1198 self._keep_client = self._api_client.keep
1199 return self._api_client
1203 if self._keep_client is None:
1204 if self._api_client is None:
1207 self._keep_client = KeepClient(api_client=self._api_client)
1208 return self._keep_client
1211 def _my_block_manager(self):
1212 if self._block_manager is None:
1213 self._block_manager = _BlockManager(self._my_keep())
1214 return self._block_manager
1216 def _populate_from_api_server(self):
1217 # As in KeepClient itself, we must wait until the last
1218 # possible moment to instantiate an API client, in order to
1219 # avoid tripping up clients that don't have access to an API
1220 # server. If we do build one, make sure our Keep client uses
1221 # it. If instantiation fails, we'll fall back to the except
1222 # clause, just like any other Collection lookup
1223 # failure. Return an exception, or None if successful.
1225 self._api_response = self._my_api().collections().get(
1226 uuid=self._manifest_locator).execute(
1227 num_retries=self.num_retries)
1228 self._manifest_text = self._api_response['manifest_text']
1230 except Exception as e:
1233 def _populate_from_keep(self):
1234 # Retrieve a manifest directly from Keep. This has a chance of
1235 # working if [a] the locator includes a permission signature
1236 # or [b] the Keep services are operating in world-readable
1237 # mode. Return an exception, or None if successful.
1239 self._manifest_text = self._my_keep().get(
1240 self._manifest_locator, num_retries=self.num_retries)
1241 except Exception as e:
1244 def _populate(self):
1245 if self._manifest_locator is None and self._manifest_text is None:
1247 error_via_api = None
1248 error_via_keep = None
1249 should_try_keep = ((self._manifest_text is None) and
1250 util.keep_locator_pattern.match(
1251 self._manifest_locator))
1252 if ((self._manifest_text is None) and
1253 util.signed_locator_pattern.match(self._manifest_locator)):
1254 error_via_keep = self._populate_from_keep()
1255 if self._manifest_text is None:
1256 error_via_api = self._populate_from_api_server()
1257 if error_via_api is not None and not should_try_keep:
1259 if ((self._manifest_text is None) and
1260 not error_via_keep and
1262 # Looks like a keep locator, and we didn't already try keep above
1263 error_via_keep = self._populate_from_keep()
1264 if self._manifest_text is None:
1266 raise errors.NotFoundError(
1267 ("Failed to retrieve collection '{}' " +
1268 "from either API server ({}) or Keep ({})."
1270 self._manifest_locator,
1274 self._baseline_manifest = self._manifest_text
1275 self._import_manifest(self._manifest_text)
1278 def _has_collection_uuid(self):
1279 return self._manifest_locator is not None and re.match(util.collection_uuid_pattern, self._manifest_locator)
1281 def __enter__(self):
1284 def __exit__(self, exc_type, exc_value, traceback):
1285 """Support scoped auto-commit in a with: block."""
1286 if exc_type is None:
1287 if self.writable() and self._has_collection_uuid():
1289 if self._block_manager is not None:
1290 self._block_manager.stop_threads()
1293 def manifest_locator(self):
1294 """Get the manifest locator, if any.
1296 The manifest locator will be set when the collection is loaded from an
1297 API server record or the portable data hash of a manifest.
1299 The manifest locator will be None if the collection is newly created or
1300 was created directly from manifest text. The method `save_new()` will
1301 assign a manifest locator.
1304 return self._manifest_locator
1307 def clone(self, new_parent=None, new_name=None, readonly=False, new_config=None):
1308 if new_config is None:
1309 new_config = self._config
1311 newcollection = CollectionReader(parent=new_parent, apiconfig=new_config)
1313 newcollection = Collection(parent=new_parent, apiconfig=new_config)
1315 newcollection._clonefrom(self)
1316 return newcollection
1319 def api_response(self):
1320 """Returns information about this Collection fetched from the API server.
1322 If the Collection exists in Keep but not the API server, currently
1323 returns None. Future versions may provide a synthetic response.
1326 return self._api_response
1328 def find_or_create(self, path, create_type):
1329 """See `RichCollectionBase.find_or_create`"""
1333 return super(Collection, self).find_or_create(path[2:] if path.startswith("./") else path, create_type)
1335 def find(self, path):
1336 """See `RichCollectionBase.find`"""
1340 return super(Collection, self).find(path[2:] if path.startswith("./") else path)
1342 def remove(self, path, recursive=False):
1343 """See `RichCollectionBase.remove`"""
1345 raise errors.ArgumentError("Cannot remove '.'")
1347 return super(Collection, self).remove(path[2:] if path.startswith("./") else path, recursive)
1352 def save(self, merge=True, num_retries=None):
1353 """Save collection to an existing collection record.
1355 Commit pending buffer blocks to Keep, merge with remote record (if
1356 merge=True, the default), and update the collection record. Returns
1357 the current manifest text.
1359 Will raise AssertionError if not associated with a collection record on
1360 the API server. If you want to save a manifest to Keep only, see
1364 Update and merge remote changes before saving. Otherwise, any
1365 remote changes will be ignored and overwritten.
1368 Retry count on API calls (if None, use the collection default)
1372 if not self._has_collection_uuid():
1373 raise AssertionError("Collection manifest_locator must be a collection uuid. Use save_new() for new collections.")
1375 self._my_block_manager().commit_all()
1380 text = self.manifest_text(strip=False)
1381 self._api_response = self._my_api().collections().update(
1382 uuid=self._manifest_locator,
1383 body={'manifest_text': text}
1385 num_retries=num_retries)
1386 self._manifest_text = self._api_response["manifest_text"]
1387 self.set_unmodified()
1389 return self._manifest_text
1395 def save_new(self, name=None,
1396 create_collection_record=True,
1398 ensure_unique_name=False,
1400 """Save collection to a new collection record.
1402 Commit pending buffer blocks to Keep and, when create_collection_record
1403 is True (default), create a new collection record. After creating a
1404 new collection record, this Collection object will be associated with
1405 the new record used by `save()`. Returns the current manifest text.
1408 The collection name.
1410 :create_collection_record:
1411 If True, create a collection record on the API server.
1412 If False, only commit blocks to Keep and return the manifest text.
1415 the user, or project uuid that will own this collection.
1416 If None, defaults to the current user.
1418 :ensure_unique_name:
1419 If True, ask the API server to rename the collection
1420 if it conflicts with a collection with the same name and owner. If
1421 False, a name conflict will result in an error.
1424 Retry count on API calls (if None, use the collection default)
1427 self._my_block_manager().commit_all()
1428 text = self.manifest_text(strip=False)
1430 if create_collection_record:
1432 name = "Collection created %s" % (time.strftime("%Y-%m-%d %H:%M:%S %Z", time.localtime()))
1433 ensure_unique_name = True
1435 body = {"manifest_text": text,
1438 body["owner_uuid"] = owner_uuid
1440 self._api_response = self._my_api().collections().create(ensure_unique_name=ensure_unique_name, body=body).execute(num_retries=num_retries)
1441 text = self._api_response["manifest_text"]
1443 self._manifest_locator = self._api_response["uuid"]
1445 self._manifest_text = text
1446 self.set_unmodified()
1451 def _import_manifest(self, manifest_text):
1452 """Import a manifest into a `Collection`.
1455 The manifest text to import from.
1459 raise ArgumentError("Can only import manifest into an empty collection")
1468 for token_and_separator in re.finditer(r'(\S+)(\s+|$)', manifest_text):
1469 tok = token_and_separator.group(1)
1470 sep = token_and_separator.group(2)
1472 if state == STREAM_NAME:
1473 # starting a new stream
1474 stream_name = tok.replace('\\040', ' ')
1479 self.mkdirs(stream_name)
1483 block_locator = re.match(r'[0-9a-f]{32}\+(\d+)(\+\S+)*', tok)
1485 blocksize = long(block_locator.group(1))
1486 blocks.append(Range(tok, streamoffset, blocksize, 0))
1487 streamoffset += blocksize
1491 if state == SEGMENTS:
1492 file_segment = re.search(r'^(\d+):(\d+):(\S+)', tok)
1494 pos = long(file_segment.group(1))
1495 size = long(file_segment.group(2))
1496 name = file_segment.group(3).replace('\\040', ' ')
1497 filepath = os.path.join(stream_name, name)
1498 afile = self.find_or_create(filepath, FILE)
1499 if isinstance(afile, ArvadosFile):
1500 afile.add_segment(blocks, pos, size)
1502 raise errors.SyntaxError("File %s conflicts with stream of the same name.", filepath)
1505 raise errors.SyntaxError("Invalid manifest format")
1511 self.set_unmodified()
1514 def notify(self, event, collection, name, item):
1516 self._callback(event, collection, name, item)
1519 class Subcollection(RichCollectionBase):
1520 """This is a subdirectory within a collection that doesn't have its own API
1523 It falls under the umbrella of the root collection.
1527 def __init__(self, parent, name):
1528 super(Subcollection, self).__init__(parent)
1529 self.lock = self.root_collection().lock
1530 self._manifest_text = None
1532 self.num_retries = parent.num_retries
1534 def root_collection(self):
1535 return self.parent.root_collection()
1538 return self.root_collection().writable()
1541 return self.root_collection()._my_api()
1544 return self.root_collection()._my_keep()
1546 def _my_block_manager(self):
1547 return self.root_collection()._my_block_manager()
1549 def stream_name(self):
1550 return os.path.join(self.parent.stream_name(), self.name)
1553 def clone(self, new_parent, new_name):
1554 c = Subcollection(new_parent, new_name)
1559 class CollectionReader(Collection):
1560 """A read-only collection object.
1562 Initialize from an api collection record locator, a portable data hash of a
1563 manifest, or raw manifest text. See `Collection` constructor for detailed
1567 def __init__(self, manifest_locator_or_text, *args, **kwargs):
1568 self._in_init = True
1569 super(CollectionReader, self).__init__(manifest_locator_or_text, *args, **kwargs)
1570 self._in_init = False
1572 # Forego any locking since it should never change once initialized.
1573 self.lock = NoopLock()
1575 # Backwards compatability with old CollectionReader
1576 # all_streams() and all_files()
1577 self._streams = None
1580 return self._in_init
1582 def _populate_streams(orig_func):
1583 @functools.wraps(orig_func)
1584 def populate_streams_wrapper(self, *args, **kwargs):
1585 # Defer populating self._streams until needed since it creates a copy of the manifest.
1586 if self._streams is None:
1587 if self._manifest_text:
1588 self._streams = [sline.split()
1589 for sline in self._manifest_text.split("\n")
1593 return orig_func(self, *args, **kwargs)
1594 return populate_streams_wrapper
1597 def normalize(self):
1598 """Normalize the streams returned by `all_streams`.
1600 This method is kept for backwards compatability and only affects the
1601 behavior of `all_streams()` and `all_files()`
1607 for s in self.all_streams():
1608 for f in s.all_files():
1609 streamname, filename = split(s.name() + "/" + f.name())
1610 if streamname not in streams:
1611 streams[streamname] = {}
1612 if filename not in streams[streamname]:
1613 streams[streamname][filename] = []
1614 for r in f.segments:
1615 streams[streamname][filename].extend(s.locators_and_ranges(r.locator, r.range_size))
1617 self._streams = [normalize_stream(s, streams[s])
1618 for s in sorted(streams)]
1620 def all_streams(self):
1621 return [StreamReader(s, self._my_keep(), num_retries=self.num_retries)
1622 for s in self._streams]
1625 def all_files(self):
1626 for s in self.all_streams():
1627 for f in s.all_files():