projects
/
arvados.git
/ blobdiff
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Merge branch '19597-multipart-request'
[arvados.git]
/
sdk
/
python
/
arvados
/
stream.py
diff --git
a/sdk/python/arvados/stream.py
b/sdk/python/arvados/stream.py
index 85c0320119acde2b745458b03d07abaa87fbbabd..edfb7711b829a100688f82bff203ebfec986096d 100644
(file)
--- a/
sdk/python/arvados/stream.py
+++ b/
sdk/python/arvados/stream.py
@@
-1,3
+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
import collections
import hashlib
import os
@@
-9,10
+17,10
@@
import copy
from ._ranges import locators_and_ranges, Range
from .arvfile import StreamFileReader
from arvados.retry import retry_method
from ._ranges import locators_and_ranges, Range
from .arvfile import StreamFileReader
from arvados.retry import retry_method
-from keep import *
-import config
-import errors
-from _normalize_stream import normalize_stream
+from
arvados.
keep import *
+
from .
import config
+
from .
import errors
+from
.
_normalize_stream import normalize_stream
class StreamReader(object):
def __init__(self, tokens, keep=None, debug=False, _empty=False,
class StreamReader(object):
def __init__(self, tokens, keep=None, debug=False, _empty=False,
@@
-23,29
+31,29
@@
class StreamReader(object):
self._keep = keep
self.num_retries = num_retries
self._keep = keep
self.num_retries = num_retries
- streamoffset = 0
L
+ streamoffset = 0
# parse stream
for tok in tokens:
# parse stream
for tok in tokens:
- if debug: print
'tok', tok
+ if debug: print
('tok', tok)
if self._stream_name is None:
self._stream_name = tok.replace('\\040', ' ')
continue
s = re.match(r'^[0-9a-f]{32}\+(\d+)(\+\S+)*$', tok)
if s:
if self._stream_name is None:
self._stream_name = tok.replace('\\040', ' ')
continue
s = re.match(r'^[0-9a-f]{32}\+(\d+)(\+\S+)*$', tok)
if s:
- blocksize =
long
(s.group(1))
- self._data_locators.append(Range(tok, streamoffset, blocksize))
+ 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:
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:
name = s.group(3).replace('\\040', ' ')
if name not in self._files:
- self._files[name] = StreamFileReader(self, [Range(pos, 0, size)], name)
+ self._files[name] = StreamFileReader(self, [Range(pos, 0, size
, 0
)], name)
else:
filereader = self._files[name]
filereader.segments.append(Range(pos, filereader.size(), size))
else:
filereader = self._files[name]
filereader.segments.append(Range(pos, filereader.size(), size))
@@
-60,15
+68,12
@@
class StreamReader(object):
return self._files
def all_files(self):
return self._files
def all_files(self):
- return
self._files.values(
)
+ return
listvalues(self._files
)
- def
_
size(self):
+ def size(self):
n = self._data_locators[-1]
return n.range_start + n.range_size
n = self._data_locators[-1]
return n.range_start + n.range_size
- def size(self):
- return self._size()
-
def locators_and_ranges(self, range_start, range_size):
return locators_and_ranges(self._data_locators, range_start, range_size)
def locators_and_ranges(self, range_start, range_size):
return locators_and_ranges(self._data_locators, range_start, range_size)
@@
-78,19
+83,15
@@
class StreamReader(object):
@retry_method
def readfrom(self, start, size, num_retries=None):
@retry_method
def readfrom(self, start, size, num_retries=None):
- return self._readfrom(start, size, num_retries=num_retries)
-
- @retry_method
- def _readfrom(self, start, size, num_retries=None):
"""Read up to 'size' bytes from the stream, starting at 'start'"""
if size == 0:
"""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])
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')]
def manifest_text(self, strip=False):
manifest_text = [self.name().replace(' ', '\\040')]
@@
-102,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])
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'
return ' '.join(manifest_text) + '\n'