X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/017587d701e08b38d9468dce45e121a69ecd8135..e158f485053be1e840073b321033d60d686a55a8:/sdk/python/arvados/collection.py diff --git a/sdk/python/arvados/collection.py b/sdk/python/arvados/collection.py index 9c536b705f..fa782a1ac2 100644 --- a/sdk/python/arvados/collection.py +++ b/sdk/python/arvados/collection.py @@ -1,28 +1,13 @@ -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 * +from .arvfile import ArvadosFileBase from keep import * -from stream import * +from .stream import StreamReader, split import config import errors import util @@ -51,7 +36,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 +45,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 +56,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 +74,17 @@ 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): @@ -174,22 +162,22 @@ class CollectionReader(CollectionBase): 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)): error_via_keep = self._populate_from_keep() - if not self._manifest_text: + if self._manifest_text is None: error_via_api = self._populate_from_api_server() - if error_via_api != None and not should_try_keep: + if error_via_api is not None and not should_try_keep: raise error_via_api - if (not self._manifest_text and + 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 not self._manifest_text: + if self._manifest_text is None: # Nothing worked! raise arvados.errors.NotFoundError( ("Failed to retrieve collection '{}' " + @@ -209,10 +197,7 @@ class CollectionReader(CollectionBase): streams = {} for s in self.all_streams(): for f in s.all_files(): - filestream = s.name() + "/" + f.name() - r = filestream.rindex("/") - streamname = filestream[:r] - filename = filestream[r+1:] + streamname, filename = split(s.name() + "/" + f.name()) if streamname not in streams: streams[streamname] = {} if filename not in streams[streamname]: @@ -228,6 +213,31 @@ class CollectionReader(CollectionBase): [StreamReader(stream, keep=self._my_keep()).manifest_text() for stream in self._streams]) + def open(self, streampath, filename=None): + """open(streampath[, filename]) -> file-like object + + Pass in the path of a file to read from the Collection, either as a + single string or as two separate stream name and file name arguments. + This method returns a file-like object to read that file. + """ + self._populate() + if filename is None: + streampath, filename = split(streampath) + keep_client = self._my_keep() + for stream_s in self._streams: + stream = StreamReader(stream_s, keep_client, + num_retries=self.num_retries) + if stream.name() == streampath: + break + else: + raise ValueError("stream '{}' not found in Collection". + format(streampath)) + try: + return stream.files()[filename] + except KeyError: + raise ValueError("file '{}' not found in Collection stream '{}'". + format(filename, streampath)) + def all_streams(self): self._populate() return [StreamReader(s, self._my_keep(), num_retries=self.num_retries) @@ -250,6 +260,29 @@ class CollectionReader(CollectionBase): return self._manifest_text +class _WriterFile(ArvadosFileBase): + def __init__(self, coll_writer, name): + super(_WriterFile, self).__init__(name, 'wb') + self.dest = coll_writer + + def close(self): + super(_WriterFile, self).close() + self.dest.finish_current_file() + + @ArvadosFileBase._before_close + def write(self, data): + self.dest.write(data) + + @ArvadosFileBase._before_close + def writelines(self, seq): + for data in seq: + self.write(data) + + @ArvadosFileBase._before_close + def flush(self): + self.dest.flush_data() + + class CollectionWriter(CollectionBase): KEEP_BLOCK_SIZE = 2**26 @@ -286,9 +319,11 @@ class CollectionWriter(CollectionBase): self._queued_file = None self._queued_dirents = deque() self._queued_trees = deque() + self._last_open = None - 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: @@ -345,10 +380,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() @@ -392,6 +426,35 @@ class CollectionWriter(CollectionBase): while self._data_buffer_len >= self.KEEP_BLOCK_SIZE: self.flush_data() + def open(self, streampath, filename=None): + """open(streampath[, filename]) -> file-like object + + Pass in the path of a file to write to the Collection, either as a + single string or as two separate stream name and file name arguments. + This method returns a file-like object you can write to add it to the + Collection. + + You may only have one file object from the Collection open at a time, + so be sure to close the object when you're done. Using the object in + a with statement makes that easy:: + + with cwriter.open('./doc/page1.txt') as outfile: + outfile.write(page1_data) + with cwriter.open('./doc/page2.txt') as outfile: + outfile.write(page2_data) + """ + if filename is None: + streampath, filename = split(streampath) + if self._last_open and not self._last_open.closed: + raise errors.AssertionError( + "can't open '{}' when '{}' is still open".format( + filename, self._last_open.name)) + if streampath != self.current_stream_name(): + self.start_new_stream(streampath) + self.set_current_file_name(filename) + self._last_open = _WriterFile(self, filename) + return self._last_open + def flush_data(self): data_buffer = ''.join(self._data_buffer) if data_buffer: @@ -415,7 +478,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( @@ -486,10 +549,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 = []