Gecko [ v3.cesa.co.za ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
addlhelp	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
commands	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
data	[ DIR ]	drwxr-xr-x	2026-06-08 18:07
tests	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
third_party	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
utils	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
vendored	[ DIR ]	drwxr-xr-x	2026-06-08 18:08
README	378 B	-rw-r--r--	1980-01-01 08:00
__init__.py	6.3 KB	-rw-r--r--	1980-01-01 08:00
__main__.py	34.63 KB	-rw-r--r--	1980-01-01 08:00
boto_resumable_upload.py	24.69 KB	-rw-r--r--	1980-01-01 08:00
boto_translation.py	85.52 KB	-rw-r--r--	1980-01-01 08:00
bucket_listing_ref.py	3.9 KB	-rw-r--r--	1980-01-01 08:00
cloud_api.py	40.97 KB	-rw-r--r--	1980-01-01 08:00
cloud_api_delegator.py	29.87 KB	-rw-r--r--	1980-01-01 08:00
command.py	109.98 KB	-rw-r--r--	1980-01-01 08:00
command_argument.py	4.35 KB	-rw-r--r--	1980-01-01 08:00
command_runner.py	23.21 KB	-rw-r--r--	1980-01-01 08:00
context_config.py	8.36 KB	-rw-r--r--	1980-01-01 08:00
cred_types.py	1.15 KB	-rw-r--r--	1980-01-01 08:00
cs_api_map.py	3.31 KB	-rw-r--r--	1980-01-01 08:00
daisy_chain_wrapper.py	14.39 KB	-rw-r--r--	1980-01-01 08:00
devshell_auth_plugin.py	1.69 KB	-rw-r--r--	1980-01-01 08:00
discard_messages_queue.py	1.29 KB	-rw-r--r--	1980-01-01 08:00
exception.py	5.18 KB	-rw-r--r--	1980-01-01 08:00
file_part.py	3.39 KB	-rw-r--r--	1980-01-01 08:00
gcs_json_api.py	104.02 KB	-rw-r--r--	1980-01-01 08:00
gcs_json_credentials.py	20.31 KB	-rw-r--r--	1980-01-01 08:00
gcs_json_media.py	28.5 KB	-rw-r--r--	1980-01-01 08:00
help_provider.py	4.26 KB	-rw-r--r--	1980-01-01 08:00
iamcredentials_api.py	11.72 KB	-rw-r--r--	1980-01-01 08:00
impersonation_credentials.py	2.51 KB	-rw-r--r--	1980-01-01 08:00
kms_api.py	13.24 KB	-rw-r--r--	1980-01-01 08:00
lazy_wrapper.py	2.26 KB	-rw-r--r--	1980-01-01 08:00
metrics.py	38.98 KB	-rw-r--r--	1980-01-01 08:00
metrics_reporter.py	4.61 KB	-rw-r--r--	1980-01-01 08:00
metrics_tuple.py	1.44 KB	-rw-r--r--	1980-01-01 08:00
name_expansion.py	29.97 KB	-rw-r--r--	1980-01-01 08:00
no_op_auth_plugin.py	1.19 KB	-rw-r--r--	1980-01-01 08:00
no_op_credentials.py	1.03 KB	-rw-r--r--	1980-01-01 08:00
parallel_tracker_file.py	12.14 KB	-rw-r--r--	1980-01-01 08:00
plurality_checkable_iterator.py	4.4 KB	-rw-r--r--	1980-01-01 08:00
progress_callback.py	6.08 KB	-rw-r--r--	1980-01-01 08:00
project_id.py	1.69 KB	-rw-r--r--	1980-01-01 08:00
pubsub_api.py	11.17 KB	-rw-r--r--	1980-01-01 08:00
resumable_streaming_upload.py	8.53 KB	-rw-r--r--	1980-01-01 08:00
seek_ahead_thread.py	5.11 KB	-rw-r--r--	1980-01-01 08:00
sig_handling.py	5.64 KB	-rw-r--r--	1980-01-01 08:00
storage_uri_builder.py	2.02 KB	-rw-r--r--	1980-01-01 08:00
storage_url.py	16.16 KB	-rw-r--r--	1980-01-01 08:00
tab_complete.py	10.77 KB	-rw-r--r--	1980-01-01 08:00
thread_message.py	13.71 KB	-rw-r--r--	1980-01-01 08:00
tracker_file.py	24.23 KB	-rw-r--r--	1980-01-01 08:00
tz_utc.py	2.2 KB	-rw-r--r--	1980-01-01 08:00
ui_controller.py	47.42 KB	-rw-r--r--	1980-01-01 08:00
wildcard_iterator.py	37.73 KB	-rw-r--r--	1980-01-01 08:00

Rename

# Copyright 2014 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Helper class for streaming resumable uploads."""

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import collections
import os

import six

from gslib.exception import CommandException
from gslib.utils.boto_util import GetJsonResumableChunkSize
from gslib.utils.constants import UTF8

class ResumableStreamingJsonUploadWrapper(object):
  """Wraps an input stream in a buffer for resumable uploads.

This class takes a non-seekable input stream, buffers it, and exposes it
  as a stream with limited seek capabilities such that it can be used in a
  resumable JSON API upload.

max_buffer_size bytes of buffering is supported.
  """

def __init__(self, stream, max_buffer_size, test_small_buffer=False):
    """Initializes the wrapper.

Args:
      stream: Input stream.
      max_buffer_size: Maximum size of internal buffer; should be >= the chunk
          size of the resumable upload API to ensure that at least one full
          chunk write can be replayed in the event of a server error.
      test_small_buffer: Skip check for buffer size vs. chunk size, for testing.
    """
    self._orig_fp = stream

if not test_small_buffer and max_buffer_size < GetJsonResumableChunkSize():
      raise CommandException('Resumable streaming upload created with buffer '
                             'size %s, JSON resumable upload chunk size %s. '
                             'Buffer size must be >= JSON resumable upload '
                             'chunk size to ensure that uploads can be '
                             'resumed.' %
                             (max_buffer_size, GetJsonResumableChunkSize()))

self._max_buffer_size = max_buffer_size
    self._buffer = collections.deque()
    self._buffer_start = 0
    self._buffer_end = 0
    self._position = 0

@property
  def mode(self):
    """Returns the mode of the underlying file descriptor, or None."""
    return getattr(self._orig_fp, 'mode', None)

def read(self, size=-1):  # pylint: disable=invalid-name
    """"Reads from the wrapped stream.

Args:
      size: The amount of bytes to read. If omitted or negative, the entire
          contents of the stream will be read and returned.

Returns:
      Bytes from the wrapped stream.
    """
    read_all_bytes = size is None or size < 0
    if read_all_bytes:
      bytes_remaining = self._max_buffer_size
    else:
      bytes_remaining = size
    data = b''
    buffered_data = []
    if self._position < self._buffer_end:
      # There was a backwards seek, so read from the buffer first.

# TODO: Performance test to validate if it is worth re-aligning
      # the buffers in this case.  Also, seeking through the buffer for
      # each read on a long catch-up is probably not performant, but we'd
      # need a more complex data structure than a deque to get around this.
      pos_in_buffer = self._buffer_start
      buffer_index = 0
      # First, find the start position in the buffer.
      while pos_in_buffer + len(self._buffer[buffer_index]) < self._position:
        # When this loop exits, buffer_index will refer to a buffer that
        # has at least some overlap with self._position, and
        # pos_in_buffer will be >= self._position
        pos_in_buffer += len(self._buffer[buffer_index])
        buffer_index += 1

# Read until we've read enough or we're out of buffer.
      while pos_in_buffer < self._buffer_end and bytes_remaining > 0:
        buffer_len = len(self._buffer[buffer_index])
        # This describes how far into the current buffer self._position is.
        offset_from_position = self._position - pos_in_buffer
        bytes_available_this_buffer = buffer_len - offset_from_position
        read_size = min(bytes_available_this_buffer, bytes_remaining)
        buffered_data.append(self._buffer[buffer_index]
                             [offset_from_position:offset_from_position +
                              read_size])
        bytes_remaining -= read_size
        pos_in_buffer += buffer_len
        buffer_index += 1
        self._position += read_size

# At this point we're guaranteed that if there are any bytes left to read,
    # then self._position == self._buffer_end, and we can read from the
    # wrapped stream if needed.
    if read_all_bytes:
      # TODO: The user is requesting reading until the end of an
      # arbitrary length stream, which is bad we'll need to return data
      # with no size limits; if the stream is sufficiently long, we could run
      # out of memory. We could break this down into smaller reads and
      # buffer it as we go, but we're still left returning the data all at
      # once to the caller.  We could raise, but for now trust the caller to
      # be sane and have enough memory to hold the remaining stream contents.
      new_data = self._orig_fp.read(size)
      data_len = len(new_data)
      if not buffered_data:
        data = new_data
      else:
        buffered_data.append(new_data)
        data = b''.join(buffered_data)
      self._position += data_len
    elif bytes_remaining:
      new_data = self._orig_fp.read(bytes_remaining)
      if not buffered_data:
        data = new_data
      else:
        buffered_data.append(new_data)
        data = b''.join(buffered_data)
      data_len = len(new_data)
      if data_len:
        self._position += data_len
        self._buffer.append(new_data)
        self._buffer_end += data_len
        oldest_data = None
        while self._buffer_end - self._buffer_start > self._max_buffer_size:
          oldest_data = self._buffer.popleft()
          self._buffer_start += len(oldest_data)
        if oldest_data:
          refill_amount = self._max_buffer_size - (self._buffer_end -
                                                   self._buffer_start)
          if refill_amount:
            self._buffer.appendleft(oldest_data[-refill_amount:])
            self._buffer_start -= refill_amount
    else:
      if six.PY3:
        if buffered_data:
          buffered_data = [
              bd.encode(UTF8) if isinstance(bd, str) else bd
              for bd in buffered_data
          ]
      data = b''.join(buffered_data) if buffered_data else b''

return data

def tell(self):  # pylint: disable=invalid-name
    """Returns the current stream position."""
    return self._position

def seekable(self):  # pylint: disable=invalid-name
    """Returns true since limited seek support exists."""
    return True

def seek(self, offset, whence=os.SEEK_SET):  # pylint: disable=invalid-name
    """Seeks on the buffered stream.

Args:
      offset: The offset to seek to; must be within the buffer bounds.
      whence: Must be os.SEEK_SET.

Raises:
      CommandException if an unsupported seek mode or position is used.
    """
    if whence == os.SEEK_SET:
      if offset < self._buffer_start or offset > self._buffer_end:
        raise CommandException('Unable to resume upload because of limited '
                               'buffering available for streaming uploads. '
                               'Offset %s was requested, but only data from '
                               '%s to %s is buffered.' %
                               (offset, self._buffer_start, self._buffer_end))
      # Move to a position within the buffer.
      self._position = offset
    elif whence == os.SEEK_END:
      if offset > self._max_buffer_size:
        raise CommandException('Invalid SEEK_END offset %s on streaming '
                               'upload. Only %s can be buffered.' %
                               (offset, self._max_buffer_size))
      # Read to the end and rely on buffering to handle the offset.
      while self.read(self._max_buffer_size):
        pass
      # Now we're at the end.
      self._position -= offset
    else:
      raise CommandException('Invalid seek mode on streaming upload. '
                             '(mode %s, offset %s)' % (whence, offset))

def close(self):  # pylint: disable=invalid-name
    return self._orig_fp.close()