X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/f69e114cf6dbe51525e98ca30a52ef3dd4341167..c8828ded72f562811dfe29ca20809dd5641f7a1d:/sdk/python/arvados/stream.py diff --git a/sdk/python/arvados/stream.py b/sdk/python/arvados/stream.py index 042ed47e40..edfb7711b8 100644 --- a/sdk/python/arvados/stream.py +++ b/sdk/python/arvados/stream.py @@ -1,5 +1,11 @@ +# Copyright (C) The Arvados Authors. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + from __future__ import print_function from __future__ import absolute_import +from future.utils import listvalues +from builtins import object import collections import hashlib import os @@ -11,7 +17,7 @@ import copy from ._ranges import locators_and_ranges, Range from .arvfile import StreamFileReader from arvados.retry import retry_method -from .keep import * +from arvados.keep import * from . import config from . import errors from ._normalize_stream import normalize_stream @@ -36,15 +42,15 @@ class StreamReader(object): s = re.match(r'^[0-9a-f]{32}\+(\d+)(\+\S+)*$', tok) if s: - blocksize = long(s.group(1)) + blocksize = int(s.group(1)) self._data_locators.append(Range(tok, streamoffset, blocksize, 0)) streamoffset += blocksize continue s = re.search(r'^(\d+):(\d+):(\S+)', tok) if s: - pos = long(s.group(1)) - size = long(s.group(2)) + pos = int(s.group(1)) + size = int(s.group(2)) name = s.group(3).replace('\\040', ' ') if name not in self._files: self._files[name] = StreamFileReader(self, [Range(pos, 0, size, 0)], name) @@ -62,7 +68,7 @@ class StreamReader(object): return self._files def all_files(self): - return self._files.values() + return listvalues(self._files) def size(self): n = self._data_locators[-1] @@ -79,13 +85,13 @@ class StreamReader(object): def readfrom(self, start, size, num_retries=None): """Read up to 'size' bytes from the stream, starting at 'start'""" if size == 0: - return '' + return b'' if self._keep is None: self._keep = KeepClient(num_retries=self.num_retries) data = [] for lr in locators_and_ranges(self._data_locators, start, size): data.append(self._keepget(lr.locator, num_retries=num_retries)[lr.segment_offset:lr.segment_offset+lr.segment_size]) - return ''.join(data) + return b''.join(data) def manifest_text(self, strip=False): manifest_text = [self.name().replace(' ', '\\040')] @@ -97,5 +103,5 @@ class StreamReader(object): manifest_text.extend([d.locator for d in self._data_locators]) manifest_text.extend([' '.join(["{}:{}:{}".format(seg.locator, seg.range_size, f.name.replace(' ', '\\040')) for seg in f.segments]) - for f in self._files.values()]) + for f in listvalues(self._files)]) return ' '.join(manifest_text) + '\n'