X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/2b6948ff7e11cd236ea26562ed76a534dcf3c0be..7924077e3db2898de26c86599d9f311e02d6db46:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index 5b00ea0a2a..59dc49fbbc 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -1,22 +1,6 @@ -import gflags -import httplib -import httplib2 import logging import os -import pprint -import sys -import types -import subprocess -import json -import UserDict import re -import hashlib -import string -import bz2 -import zlib -import fcntl -import time -import threading from collections import deque from stat import * @@ -51,7 +35,7 @@ def normalize_stream(s, stream): fout = f.replace(' ', '\\040') for segment in stream[f]: segmentoffset = blocks[segment[arvados.LOCATOR]] + segment[arvados.OFFSET] - if current_span == None: + if current_span is None: current_span = [segmentoffset, segmentoffset + segment[arvados.SEGMENTSIZE]] else: if segmentoffset == current_span[1]: @@ -60,10 +44,10 @@ def normalize_stream(s, stream): stream_tokens.append("{0}:{1}:{2}".format(current_span[0], current_span[1] - current_span[0], fout)) current_span = [segmentoffset, segmentoffset + segment[arvados.SEGMENTSIZE]] - if current_span != None: + if current_span is not None: stream_tokens.append("{0}:{1}:{2}".format(current_span[0], current_span[1] - current_span[0], fout)) - if len(stream[f]) == 0: + if not stream[f]: stream_tokens.append("0:0:{0}".format(fout)) return stream_tokens @@ -71,9 +55,9 @@ def normalize_stream(s, stream): class CollectionBase(object): def __enter__(self): - pass + return self - def __exit__(self): + def __exit__(self, exc_type, exc_value, traceback): pass def _my_keep(self): @@ -89,14 +73,18 @@ class CollectionBase(object): hints other than size hints) removed from the locators. """ raw = self.manifest_text() - clean = '' + clean = [] for line in raw.split("\n"): fields = line.split() - if len(fields) > 0: - locators = [ (re.sub(r'\+[^\d][^\+]*', '', x) if re.match(util.keep_locator_pattern, x) else x) - for x in fields[1:-1] ] - clean += fields[0] + ' ' + ' '.join(locators) + ' ' + fields[-1] + "\n" - return clean + if fields: + clean_fields = fields[:1] + [ + (re.sub(r'\+[^\d][^\+]*', '', x) + if re.match(util.keep_locator_pattern, x) + else x) + for x in fields[1:]] + clean += [' '.join(clean_fields), "\n"] + return ''.join(clean) + class CollectionReader(CollectionBase): def __init__(self, manifest_locator_or_text, api_client=None, @@ -138,58 +126,57 @@ class CollectionReader(CollectionBase): self._streams = None def _populate_from_api_server(self): - # As in KeepClient itself, we must wait until the last possible - # moment to instantiate an API client, in order to avoid - # tripping up clients that don't have access to an API server. - # If we do build one, make sure our Keep client uses it. - # If instantiation fails, we'll fall back to the except clause, - # just like any other Collection lookup failure. - if self._api_client is None: - self._api_client = arvados.api('v1') - self._keep_client = None # Make a new one with the new api. - c = self._api_client.collections().get( - uuid=self._manifest_locator).execute( - num_retries=self.num_retries) - self._manifest_text = c['manifest_text'] + # As in KeepClient itself, we must wait until the last + # possible moment to instantiate an API client, in order to + # avoid tripping up clients that don't have access to an API + # server. If we do build one, make sure our Keep client uses + # 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: + if self._api_client is None: + self._api_client = arvados.api('v1') + self._keep_client = None # Make a new one with the new api. + c = self._api_client.collections().get( + uuid=self._manifest_locator).execute( + num_retries=self.num_retries) + self._manifest_text = c['manifest_text'] + 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. - self._manifest_text = self._my_keep().get( - self._manifest_locator, num_retries=self.num_retries) + # mode. Return an exception, or None if successful. + try: + self._manifest_text = self._my_keep().get( + self._manifest_locator, num_retries=self.num_retries) + except Exception as e: + return e def _populate(self): if self._streams is not None: return error_via_api = None error_via_keep = None - should_try_keep = (not self._manifest_text and + should_try_keep = ((self._manifest_text is None) and util.keep_locator_pattern.match( self._manifest_locator)) - if (not self._manifest_text and + if ((self._manifest_text is None) and util.signed_locator_pattern.match(self._manifest_locator)): - try: - self._populate_from_keep() - except e: - error_via_keep = e - if not self._manifest_text: - try: - self._populate_from_api_server() - except Exception as e: - if not should_try_keep: - raise - error_via_api = e - if (not self._manifest_text and + 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 - try: - self._populate_from_keep() - except Exception as e: - error_via_keep = e - if not self._manifest_text: + error_via_keep = self._populate_from_keep() + if self._manifest_text is None: # Nothing worked! raise arvados.errors.NotFoundError( ("Failed to retrieve collection '{}' " + @@ -220,16 +207,13 @@ class CollectionReader(CollectionBase): for r in f.segments: streams[streamname][filename].extend(s.locators_and_ranges(r[0], r[1])) - self._streams = [] - sortedstreams = list(streams.keys()) - sortedstreams.sort() - for s in sortedstreams: - self._streams.append(normalize_stream(s, streams[s])) + self._streams = [normalize_stream(s, streams[s]) + for s in sorted(streams)] # Regenerate the manifest text based on the normalized streams - self._manifest_text = ''.join([StreamReader(stream, keep=self._my_keep()).manifest_text() for stream in self._streams]) - - return self + self._manifest_text = ''.join( + [StreamReader(stream, keep=self._my_keep()).manifest_text() + for stream in self._streams]) def all_streams(self): self._populate() @@ -241,8 +225,12 @@ class CollectionReader(CollectionBase): for f in s.all_files(): yield f - def manifest_text(self, strip=False): - if strip: + def manifest_text(self, strip=False, normalize=False): + if normalize: + cr = CollectionReader(self.manifest_text()) + cr.normalize() + return cr.manifest_text(strip=strip, normalize=False) + elif strip: return self.stripped_manifest() else: self._populate() @@ -286,8 +274,9 @@ class CollectionWriter(CollectionBase): self._queued_dirents = deque() self._queued_trees = deque() - def __exit__(self): - self.finish() + def __exit__(self, exc_type, exc_value, traceback): + if exc_type is None: + self.finish() def do_queued_work(self): # The work queue consists of three pieces: @@ -344,10 +333,9 @@ class CollectionWriter(CollectionBase): def _work_trees(self): path, stream_name, max_manifest_depth = self._queued_trees[0] - make_dirents = (util.listdir_recursive if (max_manifest_depth == 0) - else os.listdir) - d = make_dirents(path) - if len(d) > 0: + d = util.listdir_recursive( + path, max_depth = (None if max_manifest_depth == 0 else 0)) + if d: self._queue_dirents(stream_name, d) else: self._queued_trees.popleft() @@ -414,7 +402,7 @@ class CollectionWriter(CollectionBase): return self._current_file_name def finish_current_file(self): - if self._current_file_name == None: + if self._current_file_name is None: if self._current_file_pos == self._current_stream_length: return raise errors.AssertionError( @@ -485,10 +473,7 @@ class CollectionWriter(CollectionBase): manifest += ' ' + ' '.join("%d:%d:%s" % (sfile[0], sfile[1], sfile[2].replace(' ', '\\040')) for sfile in stream[2]) manifest += "\n" - if manifest: - return manifest - else: - return "" + return manifest def data_locators(self): ret = []