# Copyright 2012 Google Inc. All Rights Reserved.

"""Python wrappers for the Google Storage RESTful API."""





__all__ = ['ReadBuffer',
           'StreamingBuffer',
          ]

import collections
import os
import urlparse

from . import errors
from . import rest_api

try:
  from google.appengine.api import urlfetch
  from google.appengine.ext import ndb
except ImportError:
  from google.appengine.api import urlfetch
  from google.appengine.ext import ndb


class _StorageApi(rest_api._RestApi):
  """A simple wrapper for the Google Storage RESTful API.

  WARNING: Do NOT directly use this api. It's an implementation detail
  and is subject to change at any release.

  All async methods have similar args and returns.

  Args:
    path: The path to the Google Storage object or bucket, e.g.
      '/mybucket/myfile' or '/mybucket'.
    **kwd: Options for urlfetch. e.g.
      headers={'content-type': 'text/plain'}, payload='blah'.

  Returns:
    A ndb Future. When fulfilled, future.get_result() should return
    a tuple of (status, headers, content) that represents a HTTP response
    of Google Cloud Storage XML API.
  """

  api_url = 'https://storage.googleapis.com'
  read_only_scope = 'https://www.googleapis.com/auth/devstorage.read_only'
  read_write_scope = 'https://www.googleapis.com/auth/devstorage.read_write'
  full_control_scope = 'https://www.googleapis.com/auth/devstorage.full_control'

  def __getstate__(self):
    """Store state as part of serialization/pickling.

    Returns:
      A tuple (of dictionaries) with the state of this object
    """
    return (super(_StorageApi, self).__getstate__(), {'api_url': self.api_url})

  def __setstate__(self, state):
    """Restore state as part of deserialization/unpickling.

    Args:
      state: the tuple from a __getstate__ call
    """
    superstate, localstate = state
    super(_StorageApi, self).__setstate__(superstate)
    self.api_url = localstate['api_url']

  @ndb.tasklet
  def do_request_async(self, url, method='GET', headers=None, payload=None,
                       deadline=None, callback=None):
    """Inherit docs.

    This method translates urlfetch exceptions to more service specific ones.
    """
    if headers is None:
      headers = {}
    if 'x-goog-api-version' not in headers:
      headers['x-goog-api-version'] = '2'
    headers['accept-encoding'] = 'gzip, *'
    try:
      resp_tuple = yield super(_StorageApi, self).do_request_async(
          url, method=method, headers=headers, payload=payload,
          deadline=deadline, callback=callback)
    except urlfetch.DownloadError, e:
      raise errors.TimeoutError(
          'Request to Google Cloud Storage timed out.', e)

    raise ndb.Return(resp_tuple)


  def post_object_async(self, path, **kwds):
    """POST to an object."""
    return self.do_request_async(self.api_url + path, 'POST', **kwds)

  def put_object_async(self, path, **kwds):
    """PUT an object."""
    return self.do_request_async(self.api_url + path, 'PUT', **kwds)

  def get_object_async(self, path, **kwds):
    """GET an object.

    Note: No payload argument is supported.
    """
    return self.do_request_async(self.api_url + path, 'GET', **kwds)

  def delete_object_async(self, path, **kwds):
    """DELETE an object.

    Note: No payload argument is supported.
    """
    return self.do_request_async(self.api_url + path, 'DELETE', **kwds)

  def head_object_async(self, path, **kwds):
    """HEAD an object.

    Depending on request headers, HEAD returns various object properties,
    e.g. Content-Length, Last-Modified, and ETag.

    Note: No payload argument is supported.
    """
    return self.do_request_async(self.api_url + path, 'HEAD', **kwds)

  def get_bucket_async(self, path, **kwds):
    """GET a bucket."""
    return self.do_request_async(self.api_url + path, 'GET', **kwds)


_StorageApi = rest_api.add_sync_methods(_StorageApi)


class ReadBuffer(object):
  """A class for reading Google storage files.

  To achieve max prefetching benefit, always read by your buffer size.
  """

  DEFAULT_BUFFER_SIZE = 1024 * 1024
  MAX_REQUEST_SIZE = 30 * DEFAULT_BUFFER_SIZE

  def __init__(self,
               api,
               path,
               max_buffer_size=DEFAULT_BUFFER_SIZE,
               max_request_size=MAX_REQUEST_SIZE):
    """Constructor.

    Args:
      api: A StorageApi instance.
      path: Path to the object, e.g. '/mybucket/myfile'.
      max_buffer_size: Max bytes to buffer.
      max_request_size: Max bytes to request in one urlfetch.
    """
    self._api = api
    self._path = path
    self._max_buffer_size = max_buffer_size
    self._max_request_size = max_request_size
    self._offset = 0
    self._reset_buffer()
    self._closed = False
    self._etag = None

    self._buffer_future = self._get_segment(0, self._max_buffer_size)

    status, headers, _ = self._api.head_object(path)
    errors.check_status(status, [200])
    self._file_size = long(headers['content-length'])
    self._check_etag(headers.get('etag'))
    if self._file_size == 0:
      self._buffer_future = None

  def __getstate__(self):
    """Store state as part of serialization/pickling.

    The contents of the read buffer are not stored, only the current offset for
    data read by the client. A new read buffer is established at unpickling.
    The head information for the object (file size and etag) are stored to
    reduce startup and ensure the file has not changed.

    Returns:
      A dictionary with the state of this object
    """
    return {'api': self._api,
            'path': self._path,
            'buffer_size': self._max_buffer_size,
            'request_size': self._max_request_size,
            'etag': self._etag,
            'size': self._file_size,
            'offset': self._offset,
            'closed': self._closed}

  def __setstate__(self, state):
    """Restore state as part of deserialization/unpickling.

    Args:
      state: the dictionary from a __getstate__ call

    Along with restoring the state, pre-fetch the next read buffer.
    """
    self._api = state['api']
    self._path = state['path']
    self._max_buffer_size = state['buffer_size']
    self._max_request_size = state['request_size']
    self._etag = state['etag']
    self._file_size = state['size']
    self._offset = state['offset']
    self._reset_buffer()
    self._closed = state['closed']
    if self._offset < self._file_size and not self._closed:
      self._buffer_future = self._get_segment(self._offset,
                                              self._max_buffer_size)
    else:
      self._buffer_future = None

  def readline(self, size=-1):
    """Read one line delimited by '\n' from the file.

    A trailing newline character is kept in the string. It may be absent when a
    file ends with an incomplete line. If the size argument is non-negative,
    it specifies the maximum string size (counting the newline) to return.
    A negative size is the same as unspecified. Empty string is returned
    only when EOF is encountered immediately.

    Args:
      size: Maximum number of bytes to read. If not specified, readline stops
        only on '\n' or EOF.

    Returns:
      The data read as a string.

    Raises:
      IOError: When this buffer is closed.
    """
    self._check_open()
    self._buffer_future = None

    data_list = []

    if size == 0:
      return ''

    while True:
      if size >= 0:
        end_offset = self._buffer_offset + size
      else:
        end_offset = len(self._buffer)
      newline_offset = self._buffer.find('\n', self._buffer_offset, end_offset)

      if newline_offset >= 0:
        data_list.append(
            self._read_buffer(newline_offset + 1 - self._buffer_offset))
        return ''.join(data_list)
      else:
        result = self._read_buffer(size)
        data_list.append(result)
        size -= len(result)
        if size == 0 or self._file_size == self._offset:
          return ''.join(data_list)
        self._fill_buffer()

  def read(self, size=-1):
    """Read data from RAW file.

    Args:
      size: Number of bytes to read as integer. Actual number of bytes
        read is always equal to size unless EOF is reached. If size is
        negative or unspecified, read the entire file.

    Returns:
      data read as str.

    Raises:
      IOError: When this buffer is closed.
    """
    self._check_open()
    if self._file_size == 0:
      return ''

    if size >= 0 and size <= len(self._buffer) - self._buffer_offset:
      result = self._read_buffer(size)
    else:
      size -= len(self._buffer) - self._buffer_offset
      data_list = [self._read_buffer()]

      if self._buffer_future:
        self._reset_buffer(self._buffer_future.get_result())
        self._buffer_future = None

      if size >= 0 and size <= len(self._buffer) - self._buffer_offset:
        data_list.append(self._read_buffer(size))
      else:
        size -= len(self._buffer)
        data_list.append(self._read_buffer())
        if self._offset == self._file_size:
          return ''.join(data_list)

        if size < 0 or size >= self._file_size - self._offset:
          needs = self._file_size - self._offset
        else:
          needs = size
        data_list.extend(self._get_segments(self._offset, needs))
        self._offset += needs
      result = ''.join(data_list)
      data_list = None

    assert self._buffer_future is None
    if self._offset != self._file_size and not self._buffer:
      self._buffer_future = self._get_segment(self._offset,
                                              self._max_buffer_size)
    return result

  def _read_buffer(self, size=-1):
    """Returns bytes from self._buffer and update related offsets.

    Args:
      size: number of bytes to read. Read the entire buffer if negative.

    Returns:
      Requested bytes from buffer.
    """
    if size < 0:
      size = len(self._buffer) - self._buffer_offset
    result = self._buffer[self._buffer_offset : self._buffer_offset+size]
    self._offset += len(result)
    self._buffer_offset += len(result)
    if self._buffer_offset == len(self._buffer):
      self._reset_buffer()
    return result

  def _fill_buffer(self):
    """Fill self._buffer."""
    segments = self._get_segments(self._offset,
                                  min(self._max_buffer_size,
                                      self._max_request_size,
                                      self._file_size-self._offset))

    self._reset_buffer(''.join(segments))

  def _get_segments(self, start, request_size):
    """Get segments of the file from Google Storage as a list.

    A large request is broken into segments to avoid hitting urlfetch
    response size limit. Each segment is returned from a separate urlfetch.

    Args:
      start: start offset to request. Inclusive. Have to be within the
        range of the file.
      request_size: number of bytes to request. Can not exceed the logical
        range of the file.

    Returns:
      A list of file segments in order
    """
    end = start + request_size
    futures = []

    while request_size > self._max_request_size:
      futures.append(self._get_segment(start, self._max_request_size))
      request_size -= self._max_request_size
      start += self._max_request_size
    if start < end:
      futures.append(self._get_segment(start, end-start))
    return [fut.get_result() for fut in futures]

  @ndb.tasklet
  def _get_segment(self, start, request_size):
    """Get a segment of the file from Google Storage.

    Args:
      start: start offset of the segment. Inclusive. Have to be within the
        range of the file.
      request_size: number of bytes to request. Have to be within the range
        of the file.

    Yields:
      a segment [start, start + request_size) of the file.

    Raises:
      ValueError: if the file has changed while reading.
    """
    end = start + request_size - 1
    content_range = '%d-%d' % (start, end)
    headers = {'Range': 'bytes=' + content_range}
    status, headers, content = yield self._api.get_object_async(self._path,
                                                                headers=headers)
    errors.check_status(status, [200, 206], headers)
    self._check_etag(headers.get('etag'))
    raise ndb.Return(content)

  def _check_etag(self, etag):
    """Check if etag is the same across requests to GCS.

    If self._etag is None, set it. If etag is set, check that the new
    etag equals the old one.

    In the __init__ method, we fire one HEAD and one GET request using
    ndb tasklet. One of them would return first and set the first value.

    Args:
      etag: etag from a GCS HTTP response. None if etag is not part of the
        response header. It could be None for example in the case of GCS
        composite file.

    Raises:
      ValueError: if two etags are not equal.
    """
    if etag is None:
      return
    elif self._etag is None:
      self._etag = etag
    elif self._etag != etag:
      raise ValueError('File on GCS has changed while reading.')

  def close(self):
    self._closed = True
    self._reset_buffer()
    self._buffer_future = None

  def __enter__(self):
    return self

  def __exit__(self, atype, value, traceback):
    self.close()
    return False

  def seek(self, offset, whence=os.SEEK_SET):
    """Set the file's current offset.

    Note if the new offset is out of bound, it is adjusted to either 0 or EOF.

    Args:
      offset: seek offset as number.
      whence: seek mode. Supported modes are os.SEEK_SET (absolute seek),
        os.SEEK_CUR (seek relative to the current position), and os.SEEK_END
        (seek relative to the end, offset should be negative).

    Raises:
      IOError: When this buffer is closed.
      ValueError: When whence is invalid.
    """
    self._check_open()

    self._reset_buffer()
    self._buffer_future = None

    if whence == os.SEEK_SET:
      self._offset = offset
    elif whence == os.SEEK_CUR:
      self._offset += offset
    elif whence == os.SEEK_END:
      self._offset = self._file_size + offset
    else:
      raise ValueError('Whence mode %s is invalid.' % str(whence))

    self._offset = min(self._offset, self._file_size)
    self._offset = max(self._offset, 0)
    if self._offset != self._file_size:
      self._buffer_future = self._get_segment(self._offset,
                                              self._max_buffer_size)

  def tell(self):
    """Tell the file's current offset.

    Returns:
      current offset in reading this file.

    Raises:
      IOError: When this buffer is closed.
    """
    self._check_open()
    return self._offset

  def _check_open(self):
    if self._closed:
      raise IOError('Buffer is closed.')

  def _reset_buffer(self, new_buffer='', buffer_offset=0):
    self._buffer = new_buffer
    self._buffer_offset = buffer_offset


class StreamingBuffer(object):
  """A class for creating large objects using the 'resumable' API.

  The API is a subset of the Python writable stream API sufficient to
  support writing zip files using the zipfile module.

  The exact sequence of calls and use of headers is documented at
  https://developers.google.com/storage/docs/developer-guide#unknownresumables
  """

  _blocksize = 256 * 1024

  _maxrequestsize = 16 * _blocksize

  def __init__(self,
               api,
               path,
               content_type=None,
               gcs_headers=None):
    """Constructor.

    Args:
      api: A StorageApi instance.
      path: Path to the object, e.g. '/mybucket/myfile'.
      content_type: Optional content-type; Default value is
        delegate to Google Cloud Storage.
      gcs_headers: additional gs headers as a str->str dict, e.g
        {'x-goog-acl': 'private', 'x-goog-meta-foo': 'foo'}.
    """
    assert self._maxrequestsize > self._blocksize
    assert self._maxrequestsize % self._blocksize == 0

    self._api = api
    self._path = path

    self._buffer = collections.deque()
    self._buffered = 0
    self._written = 0
    self._offset = 0

    self._closed = False

    headers = {'x-goog-resumable': 'start'}
    if content_type:
      headers['content-type'] = content_type
    if gcs_headers:
      headers.update(gcs_headers)
    status, headers, _ = self._api.post_object(path, headers=headers)
    errors.check_status(status, [201], headers)
    loc = headers.get('location')
    if not loc:
      raise IOError('No location header found in 201 response')
    parsed = urlparse.urlparse(loc)
    self._path_with_token = '%s?%s' % (self._path, parsed.query)

  def __getstate__(self):
    """Store state as part of serialization/pickling.

    The contents of the write buffer are stored. Writes to the underlying
    storage are required to be on block boundaries (_blocksize) except for the
    last write. In the worst case the pickled version of this object may be
    slightly larger than the blocksize.

    Returns:
      A dictionary with the state of this object

    """
    return {'api': self._api,
            'path_token': self._path_with_token,
            'buffer': self._buffer,
            'buffered': self._buffered,
            'written': self._written,
            'offset': self._offset,
            'closed': self._closed}

  def __setstate__(self, state):
    """Restore state as part of deserialization/unpickling.

    Args:
      state: the dictionary from a __getstate__ call
    """
    self._api = state['api']
    self._path_with_token = state['path_token']
    self._buffer = state['buffer']
    self._buffered = state['buffered']
    self._written = state['written']
    self._offset = state['offset']
    self._closed = state['closed']

  def write(self, data):
    """Write some bytes."""
    self._check_open()
    assert isinstance(data, str)
    if not data:
      return
    self._buffer.append(data)
    self._buffered += len(data)
    self._offset += len(data)
    if self._buffered >= self._blocksize:
      self._flush()

  def flush(self):
    """Dummy API.

    This API is provided because the zipfile module uses it.  It is a
    no-op because Google Storage *requires* that all writes except for
    the final one are multiples on 256K bytes aligned on 256K-byte
    boundaries.
    """
    self._check_open()

  def tell(self):
    """Return the total number of bytes passed to write() so far.

    (There is no seek() method.)
    """
    self._check_open()
    return self._offset

  def close(self):
    """Flush the buffer and finalize the file.

    When this returns the new file is available for reading.
    """
    if not self._closed:
      self._closed = True
      self._flush(finish=True)
      self._buffer = None

  def __enter__(self):
    return self

  def __exit__(self, atype, value, traceback):
    self.close()
    return False

  def _flush(self, finish=False):
    """Internal API to flush.

    This is called only when the total amount of buffered data is at
    least self._blocksize, or to flush the final (incomplete) block of
    the file with finish=True.
    """
    flush_len = 0 if finish else self._blocksize
    last = False

    while self._buffered >= flush_len:
      buffer = []
      buffered = 0

      while self._buffer:
        buf = self._buffer.popleft()
        size = len(buf)
        self._buffered -= size
        buffer.append(buf)
        buffered += size
        if buffered >= self._maxrequestsize:
          break

      if buffered > self._maxrequestsize:
        excess = buffered - self._maxrequestsize
      elif finish:
        excess = 0
      else:
        excess = buffered % self._blocksize

      if excess:
        over = buffer.pop()
        size = len(over)
        assert size >= excess
        buffered -= size
        head, tail = over[:-excess], over[-excess:]
        self._buffer.appendleft(tail)
        self._buffered += len(tail)
        if head:
          buffer.append(head)
          buffered += len(head)

      if finish:
        last = not self._buffered
      self._send_data(''.join(buffer), last)
      if last:
        break

  def _send_data(self, data, last):
    """Send the block to the storage service and update self._written."""
    headers = {}
    length = self._written + len(data)

    if data:
      headers['content-range'] = ('bytes %d-%d/%s' %
                                  (self._written, length-1,
                                   length if last else '*'))
    else:
      headers['content-range'] = ('bytes */%s' %
                                  length if last else '*')
    status, _, _ = self._api.put_object(
        self._path_with_token, payload=data, headers=headers)
    if last:
      expected = 200
    else:
      expected = 308
    errors.check_status(status, [expected], headers)
    self._written += len(data)

  def _check_open(self):
    if self._closed:
      raise IOError('Buffer is closed.')

# Copyright 2012 Google Inc. All Rights Reserved.

"""Python wrappers for the Google Storage RESTful API."""





__all__ = ['ReadBuffer',
           'StreamingBuffer',
          ]

import collections
import os
import urlparse

from . import errors
from . import rest_api

try:
  from google.appengine.api import urlfetch
  from google.appengine.ext import ndb
except ImportError:
  from google.appengine.api import urlfetch
  from google.appengine.ext import ndb


class _StorageApi(rest_api._RestApi):
  """A simple wrapper for the Google Storage RESTful API.

  WARNING: Do NOT directly use this api. It's an implementation detail
  and is subject to change at any release.

  All async methods have similar args and returns.

  Args:
    path: The path to the Google Storage object or bucket, e.g.
      '/mybucket/myfile' or '/mybucket'.
    **kwd: Options for urlfetch. e.g.
      headers={'content-type': 'text/plain'}, payload='blah'.

  Returns:
    A ndb Future. When fulfilled, future.get_result() should return
    a tuple of (status, headers, content) that represents a HTTP response
    of Google Cloud Storage XML API.
  """

  api_url = 'https://storage.googleapis.com'
  read_only_scope = 'https://www.googleapis.com/auth/devstorage.read_only'
  read_write_scope = 'https://www.googleapis.com/auth/devstorage.read_write'
  full_control_scope = 'https://www.googleapis.com/auth/devstorage.full_control'

  def __getstate__(self):
    """Store state as part of serialization/pickling.

    Returns:
      A tuple (of dictionaries) with the state of this object
    """
    return (super(_StorageApi, self).__getstate__(), {'api_url': self.api_url})

  def __setstate__(self, state):
    """Restore state as part of deserialization/unpickling.

    Args:
      state: the tuple from a __getstate__ call
    """
    superstate, localstate = state
    super(_StorageApi, self).__setstate__(superstate)
    self.api_url = localstate['api_url']

  @ndb.tasklet
  def do_request_async(self, url, method='GET', headers=None, payload=None,
                       deadline=None, callback=None):
    """Inherit docs.

    This method translates urlfetch exceptions to more service specific ones.
    """
    if headers is None:
      headers = {}
    if 'x-goog-api-version' not in headers:
      headers['x-goog-api-version'] = '2'
    headers['accept-encoding'] = 'gzip, *'
    try:
      resp_tuple = yield super(_StorageApi, self).do_request_async(
          url, method=method, headers=headers, payload=payload,
          deadline=deadline, callback=callback)
    except urlfetch.DownloadError, e:
      raise errors.TimeoutError(
          'Request to Google Cloud Storage timed out.', e)

    raise ndb.Return(resp_tuple)


  def post_object_async(self, path, **kwds):
    """POST to an object."""
    return self.do_request_async(self.api_url + path, 'POST', **kwds)

  def put_object_async(self, path, **kwds):
    """PUT an object."""
    return self.do_request_async(self.api_url + path, 'PUT', **kwds)

  def get_object_async(self, path, **kwds):
    """GET an object.

    Note: No payload argument is supported.
    """
    return self.do_request_async(self.api_url + path, 'GET', **kwds)

  def delete_object_async(self, path, **kwds):
    """DELETE an object.

    Note: No payload argument is supported.
    """
    return self.do_request_async(self.api_url + path, 'DELETE', **kwds)

  def head_object_async(self, path, **kwds):
    """HEAD an object.

    Depending on request headers, HEAD returns various object properties,
    e.g. Content-Length, Last-Modified, and ETag.

    Note: No payload argument is supported.
    """
    return self.do_request_async(self.api_url + path, 'HEAD', **kwds)

  def get_bucket_async(self, path, **kwds):
    """GET a bucket."""
    return self.do_request_async(self.api_url + path, 'GET', **kwds)


_StorageApi = rest_api.add_sync_methods(_StorageApi)


class ReadBuffer(object):
  """A class for reading Google storage files.

  To achieve max prefetching benefit, always read by your buffer size.
  """

  DEFAULT_BUFFER_SIZE = 1024 * 1024
  MAX_REQUEST_SIZE = 30 * DEFAULT_BUFFER_SIZE

  def __init__(self,
               api,
               path,
               max_buffer_size=DEFAULT_BUFFER_SIZE,
               max_request_size=MAX_REQUEST_SIZE):
    """Constructor.

    Args:
      api: A StorageApi instance.
      path: Path to the object, e.g. '/mybucket/myfile'.
      max_buffer_size: Max bytes to buffer.
      max_request_size: Max bytes to request in one urlfetch.
    """
    self._api = api
    self._path = path
    self._max_buffer_size = max_buffer_size
    self._max_request_size = max_request_size
    self._offset = 0
    self._reset_buffer()
    self._closed = False
    self._etag = None

    self._buffer_future = self._get_segment(0, self._max_buffer_size)

    status, headers, _ = self._api.head_object(path)
    errors.check_status(status, [200])
    self._file_size = long(headers['content-length'])
    self._check_etag(headers.get('etag'))
    if self._file_size == 0:
      self._buffer_future = None

  def __getstate__(self):
    """Store state as part of serialization/pickling.

    The contents of the read buffer are not stored, only the current offset for
    data read by the client. A new read buffer is established at unpickling.
    The head information for the object (file size and etag) are stored to
    reduce startup and ensure the file has not changed.

    Returns:
      A dictionary with the state of this object
    """
    return {'api': self._api,
            'path': self._path,
            'buffer_size': self._max_buffer_size,
            'request_size': self._max_request_size,
            'etag': self._etag,
            'size': self._file_size,
            'offset': self._offset,
            'closed': self._closed}

  def __setstate__(self, state):
    """Restore state as part of deserialization/unpickling.

    Args:
      state: the dictionary from a __getstate__ call

    Along with restoring the state, pre-fetch the next read buffer.
    """
    self._api = state['api']
    self._path = state['path']
    self._max_buffer_size = state['buffer_size']
    self._max_request_size = state['request_size']
    self._etag = state['etag']
    self._file_size = state['size']
    self._offset = state['offset']
    self._reset_buffer()
    self._closed = state['closed']
    if self._offset < self._file_size and not self._closed:
      self._buffer_future = self._get_segment(self._offset,
                                              self._max_buffer_size)
    else:
      self._buffer_future = None

  def readline(self, size=-1):
    """Read one line delimited by '\n' from the file.

    A trailing newline character is kept in the string. It may be absent when a
    file ends with an incomplete line. If the size argument is non-negative,
    it specifies the maximum string size (counting the newline) to return.
    A negative size is the same as unspecified. Empty string is returned
    only when EOF is encountered immediately.

    Args:
      size: Maximum number of bytes to read. If not specified, readline stops
        only on '\n' or EOF.

    Returns:
      The data read as a string.

    Raises:
      IOError: When this buffer is closed.
    """
    self._check_open()
    self._buffer_future = None

    data_list = []

    if size == 0:
      return ''

    while True:
      if size >= 0:
        end_offset = self._buffer_offset + size
      else:
        end_offset = len(self._buffer)
      newline_offset = self._buffer.find('\n', self._buffer_offset, end_offset)

      if newline_offset >= 0:
        data_list.append(
            self._read_buffer(newline_offset + 1 - self._buffer_offset))
        return ''.join(data_list)
      else:
        result = self._read_buffer(size)
        data_list.append(result)
        size -= len(result)
        if size == 0 or self._file_size == self._offset:
          return ''.join(data_list)
        self._fill_buffer()

  def read(self, size=-1):
    """Read data from RAW file.

    Args:
      size: Number of bytes to read as integer. Actual number of bytes
        read is always equal to size unless EOF is reached. If size is
        negative or unspecified, read the entire file.

    Returns:
      data read as str.

    Raises:
      IOError: When this buffer is closed.
    """
    self._check_open()
    if self._file_size == 0:
      return ''

    if size >= 0 and size <= len(self._buffer) - self._buffer_offset:
      result = self._read_buffer(size)
    else:
      size -= len(self._buffer) - self._buffer_offset
      data_list = [self._read_buffer()]

      if self._buffer_future:
        self._reset_buffer(self._buffer_future.get_result())
        self._buffer_future = None

      if size >= 0 and size <= len(self._buffer) - self._buffer_offset:
        data_list.append(self._read_buffer(size))
      else:
        size -= len(self._buffer)
        data_list.append(self._read_buffer())
        if self._offset == self._file_size:
          return ''.join(data_list)

        if size < 0 or size >= self._file_size - self._offset:
          needs = self._file_size - self._offset
        else:
          needs = size
        data_list.extend(self._get_segments(self._offset, needs))
        self._offset += needs
      result = ''.join(data_list)
      data_list = None

    assert self._buffer_future is None
    if self._offset != self._file_size and not self._buffer:
      self._buffer_future = self._get_segment(self._offset,
                                              self._max_buffer_size)
    return result

  def _read_buffer(self, size=-1):
    """Returns bytes from self._buffer and update related offsets.

    Args:
      size: number of bytes to read. Read the entire buffer if negative.

    Returns:
      Requested bytes from buffer.
    """
    if size < 0:
      size = len(self._buffer) - self._buffer_offset
    result = self._buffer[self._buffer_offset : self._buffer_offset+size]
    self._offset += len(result)
    self._buffer_offset += len(result)
    if self._buffer_offset == len(self._buffer):
      self._reset_buffer()
    return result

  def _fill_buffer(self):
    """Fill self._buffer."""
    segments = self._get_segments(self._offset,
                                  min(self._max_buffer_size,
                                      self._max_request_size,
                                      self._file_size-self._offset))

    self._reset_buffer(''.join(segments))

  def _get_segments(self, start, request_size):
    """Get segments of the file from Google Storage as a list.

    A large request is broken into segments to avoid hitting urlfetch
    response size limit. Each segment is returned from a separate urlfetch.

    Args:
      start: start offset to request. Inclusive. Have to be within the
        range of the file.
      request_size: number of bytes to request. Can not exceed the logical
        range of the file.

    Returns:
      A list of file segments in order
    """
    end = start + request_size
    futures = []

    while request_size > self._max_request_size:
      futures.append(self._get_segment(start, self._max_request_size))
      request_size -= self._max_request_size
      start += self._max_request_size
    if start < end:
      futures.append(self._get_segment(start, end-start))
    return [fut.get_result() for fut in futures]

  @ndb.tasklet
  def _get_segment(self, start, request_size):
    """Get a segment of the file from Google Storage.

    Args:
      start: start offset of the segment. Inclusive. Have to be within the
        range of the file.
      request_size: number of bytes to request. Have to be within the range
        of the file.

    Yields:
      a segment [start, start + request_size) of the file.

    Raises:
      ValueError: if the file has changed while reading.
    """
    end = start + request_size - 1
    content_range = '%d-%d' % (start, end)
    headers = {'Range': 'bytes=' + content_range}
    status, headers, content = yield self._api.get_object_async(self._path,
                                                                headers=headers)
    errors.check_status(status, [200, 206], headers)
    self._check_etag(headers.get('etag'))
    raise ndb.Return(content)

  def _check_etag(self, etag):
    """Check if etag is the same across requests to GCS.

    If self._etag is None, set it. If etag is set, check that the new
    etag equals the old one.

    In the __init__ method, we fire one HEAD and one GET request using
    ndb tasklet. One of them would return first and set the first value.

    Args:
      etag: etag from a GCS HTTP response. None if etag is not part of the
        response header. It could be None for example in the case of GCS
        composite file.

    Raises:
      ValueError: if two etags are not equal.
    """
    if etag is None:
      return
    elif self._etag is None:
      self._etag = etag
    elif self._etag != etag:
      raise ValueError('File on GCS has changed while reading.')

  def close(self):
    self._closed = True
    self._reset_buffer()
    self._buffer_future = None

  def __enter__(self):
    return self

  def __exit__(self, atype, value, traceback):
    self.close()
    return False

  def seek(self, offset, whence=os.SEEK_SET):
    """Set the file's current offset.

    Note if the new offset is out of bound, it is adjusted to either 0 or EOF.

    Args:
      offset: seek offset as number.
      whence: seek mode. Supported modes are os.SEEK_SET (absolute seek),
        os.SEEK_CUR (seek relative to the current position), and os.SEEK_END
        (seek relative to the end, offset should be negative).

    Raises:
      IOError: When this buffer is closed.
      ValueError: When whence is invalid.
    """
    self._check_open()

    self._reset_buffer()
    self._buffer_future = None

    if whence == os.SEEK_SET:
      self._offset = offset
    elif whence == os.SEEK_CUR:
      self._offset += offset
    elif whence == os.SEEK_END:
      self._offset = self._file_size + offset
    else:
      raise ValueError('Whence mode %s is invalid.' % str(whence))

    self._offset = min(self._offset, self._file_size)
    self._offset = max(self._offset, 0)
    if self._offset != self._file_size:
      self._buffer_future = self._get_segment(self._offset,
                                              self._max_buffer_size)

  def tell(self):
    """Tell the file's current offset.

    Returns:
      current offset in reading this file.

    Raises:
      IOError: When this buffer is closed.
    """
    self._check_open()
    return self._offset

  def _check_open(self):
    if self._closed:
      raise IOError('Buffer is closed.')

  def _reset_buffer(self, new_buffer='', buffer_offset=0):
    self._buffer = new_buffer
    self._buffer_offset = buffer_offset


class StreamingBuffer(object):
  """A class for creating large objects using the 'resumable' API.

  The API is a subset of the Python writable stream API sufficient to
  support writing zip files using the zipfile module.

  The exact sequence of calls and use of headers is documented at
  https://developers.google.com/storage/docs/developer-guide#unknownresumables
  """

  _blocksize = 256 * 1024

  _maxrequestsize = 16 * _blocksize

  def __init__(self,
               api,
               path,
               content_type=None,
               gcs_headers=None):
    """Constructor.

    Args:
      api: A StorageApi instance.
      path: Path to the object, e.g. '/mybucket/myfile'.
      content_type: Optional content-type; Default value is
        delegate to Google Cloud Storage.
      gcs_headers: additional gs headers as a str->str dict, e.g
        {'x-goog-acl': 'private', 'x-goog-meta-foo': 'foo'}.
    """
    assert self._maxrequestsize > self._blocksize
    assert self._maxrequestsize % self._blocksize == 0

    self._api = api
    self._path = path

    self._buffer = collections.deque()
    self._buffered = 0
    self._written = 0
    self._offset = 0

    self._closed = False

    headers = {'x-goog-resumable': 'start'}
    if content_type:
      headers['content-type'] = content_type
    if gcs_headers:
      headers.update(gcs_headers)
    status, headers, _ = self._api.post_object(path, headers=headers)
    errors.check_status(status, [201], headers)
    loc = headers.get('location')
    if not loc:
      raise IOError('No location header found in 201 response')
    parsed = urlparse.urlparse(loc)
    self._path_with_token = '%s?%s' % (self._path, parsed.query)

  def __getstate__(self):
    """Store state as part of serialization/pickling.

    The contents of the write buffer are stored. Writes to the underlying
    storage are required to be on block boundaries (_blocksize) except for the
    last write. In the worst case the pickled version of this object may be
    slightly larger than the blocksize.

    Returns:
      A dictionary with the state of this object

    """
    return {'api': self._api,
            'path_token': self._path_with_token,
            'buffer': self._buffer,
            'buffered': self._buffered,
            'written': self._written,
            'offset': self._offset,
            'closed': self._closed}

  def __setstate__(self, state):
    """Restore state as part of deserialization/unpickling.

    Args:
      state: the dictionary from a __getstate__ call
    """
    self._api = state['api']
    self._path_with_token = state['path_token']
    self._buffer = state['buffer']
    self._buffered = state['buffered']
    self._written = state['written']
    self._offset = state['offset']
    self._closed = state['closed']

  def write(self, data):
    """Write some bytes."""
    self._check_open()
    assert isinstance(data, str)
    if not data:
      return
    self._buffer.append(data)
    self._buffered += len(data)
    self._offset += len(data)
    if self._buffered >= self._blocksize:
      self._flush()

  def flush(self):
    """Dummy API.

    This API is provided because the zipfile module uses it.  It is a
    no-op because Google Storage *requires* that all writes except for
    the final one are multiples on 256K bytes aligned on 256K-byte
    boundaries.
    """
    self._check_open()

  def tell(self):
    """Return the total number of bytes passed to write() so far.

    (There is no seek() method.)
    """
    self._check_open()
    return self._offset

  def close(self):
    """Flush the buffer and finalize the file.

    When this returns the new file is available for reading.
    """
    if not self._closed:
      self._closed = True
      self._flush(finish=True)
      self._buffer = None

  def __enter__(self):
    return self

  def __exit__(self, atype, value, traceback):
    self.close()
    return False

  def _flush(self, finish=False):
    """Internal API to flush.

    This is called only when the total amount of buffered data is at
    least self._blocksize, or to flush the final (incomplete) block of
    the file with finish=True.
    """
    flush_len = 0 if finish else self._blocksize
    last = False

    while self._buffered >= flush_len:
      buffer = []
      buffered = 0

      while self._buffer:
        buf = self._buffer.popleft()
        size = len(buf)
        self._buffered -= size
        buffer.append(buf)
        buffered += size
        if buffered >= self._maxrequestsize:
          break

      if buffered > self._maxrequestsize:
        excess = buffered - self._maxrequestsize
      elif finish:
        excess = 0
      else:
        excess = buffered % self._blocksize

      if excess:
        over = buffer.pop()
        size = len(over)
        assert size >= excess
        buffered -= size
        head, tail = over[:-excess], over[-excess:]
        self._buffer.appendleft(tail)
        self._buffered += len(tail)
        if head:
          buffer.append(head)
          buffered += len(head)

      if finish:
        last = not self._buffered
      self._send_data(''.join(buffer), last)
      if last:
        break

  def _send_data(self, data, last):
    """Send the block to the storage service and update self._written."""
    headers = {}
    length = self._written + len(data)

    if data:
      headers['content-range'] = ('bytes %d-%d/%s' %
                                  (self._written, length-1,
                                   length if last else '*'))
    else:
      headers['content-range'] = ('bytes */%s' %
                                  length if last else '*')
    status, _, _ = self._api.put_object(
        self._path_with_token, payload=data, headers=headers)
    if last:
      expected = 200
    else:
      expected = 308
    errors.check_status(status, [expected], headers)
    self._written += len(data)

  def _check_open(self):
    if self._closed:
      raise IOError('Buffer is closed.')

# Copyright 2012 Google Inc. All Rights Reserved. Python wrappers for the Google Storage RESTful API. A simple wrapper for the Google Storage RESTful API.    WARNING: Do NOT directly use this api. It's an implementation detail   and is subject to change at any release.    All async methods have similar args and returns.    Args:     path: The path to the Google Storage object or bucket, e.g.       '/mybucket/myfile' or '/mybucket'.     **kwd: Options for urlfetch. e.g.       headers={'content-type': 'text/plain'}, payload='blah'.    Returns:     A ndb Future. When fulfilled, future.get_result() should return     a tuple of (status, headers, content) that represents a HTTP response     of Google Cloud Storage XML API. Store state as part of serialization/pickling.      Returns:       A tuple (of dictionaries) with the state of this object Restore state as part of deserialization/unpickling.      Args:       state: the tuple from a __getstate__ call Inherit docs.      This method translates urlfetch exceptions to more service specific ones. POST to an object. PUT an object. GET an object.      Note: No payload argument is supported. DELETE an object.      Note: No payload argument is supported. HEAD an object.      Depending on request headers, HEAD returns various object properties,     e.g. Content-Length, Last-Modified, and ETag.      Note: No payload argument is supported. GET a bucket. A class for reading Google storage files.    To achieve max prefetching benefit, always read by your buffer size. Constructor.      Args:       api: A StorageApi instance.       path: Path to the object, e.g. '/mybucket/myfile'.       max_buffer_size: Max bytes to buffer.       max_request_size: Max bytes to request in one urlfetch. Store state as part of serialization/pickling.      The contents of the read buffer are not stored, only the current offset for     data read by the client. A new read buffer is established at unpickling.     The head information for the object (file size and etag) are stored to     reduce startup and ensure the file has not changed.      Returns:       A dictionary with the state of this object Restore state as part of deserialization/unpickling.      Args:       state: the dictionary from a __getstate__ call      Along with restoring the state, pre-fetch the next read buffer. Read one line delimited by '\n' from the file.      A trailing newline character is kept in the string. It may be absent when a     file ends with an incomplete line. If the size argument is non-negative,     it specifies the maximum string size (counting the newline) to return.     A negative size is the same as unspecified. Empty string is returned     only when EOF is encountered immediately.      Args:       size: Maximum number of bytes to read. If not specified, readline stops         only on '\n' or EOF.      Returns:       The data read as a string.      Raises:       IOError: When this buffer is closed. Read data from RAW file.      Args:       size: Number of bytes to read as integer. Actual number of bytes         read is always equal to size unless EOF is reached. If size is         negative or unspecified, read the entire file.      Returns:       data read as str.      Raises:       IOError: When this buffer is closed. Returns bytes from self._buffer and update related offsets.      Args:       size: number of bytes to read. Read the entire buffer if negative.      Returns:       Requested bytes from buffer. Fill self._buffer. Get segments of the file from Google Storage as a list.      A large request is broken into segments to avoid hitting urlfetch     response size limit. Each segment is returned from a separate urlfetch.      Args:       start: start offset to request. Inclusive. Have to be within the         range of the file.       request_size: number of bytes to request. Can not exceed the logical         range of the file.      Returns:       A list of file segments in order Get a segment of the file from Google Storage.      Args:       start: start offset of the segment. Inclusive. Have to be within the         range of the file.       request_size: number of bytes to request. Have to be within the range         of the file.      Yields:       a segment [start, start + request_size) of the file.      Raises:       ValueError: if the file has changed while reading. Check if etag is the same across requests to GCS.      If self._etag is None, set it. If etag is set, check that the new     etag equals the old one.      In the __init__ method, we fire one HEAD and one GET request using     ndb tasklet. One of them would return first and set the first value.      Args:       etag: etag from a GCS HTTP response. None if etag is not part of the         response header. It could be None for example in the case of GCS         composite file.      Raises:       ValueError: if two etags are not equal. Set the file's current offset.      Note if the new offset is out of bound, it is adjusted to either 0 or EOF.      Args:       offset: seek offset as number.       whence: seek mode. Supported modes are os.SEEK_SET (absolute seek),         os.SEEK_CUR (seek relative to the current position), and os.SEEK_END         (seek relative to the end, offset should be negative).      Raises:       IOError: When this buffer is closed.       ValueError: When whence is invalid. Tell the file's current offset.      Returns:       current offset in reading this file.      Raises:       IOError: When this buffer is closed. A class for creating large objects using the 'resumable' API.    The API is a subset of the Python writable stream API sufficient to   support writing zip files using the zipfile module.    The exact sequence of calls and use of headers is documented at   https://developers.google.com/storage/docs/developer-guide#unknownresumables Constructor.      Args:       api: A StorageApi instance.       path: Path to the object, e.g. '/mybucket/myfile'.       content_type: Optional content-type; Default value is         delegate to Google Cloud Storage.       gcs_headers: additional gs headers as a str->str dict, e.g         {'x-goog-acl': 'private', 'x-goog-meta-foo': 'foo'}. Store state as part of serialization/pickling.      The contents of the write buffer are stored. Writes to the underlying     storage are required to be on block boundaries (_blocksize) except for the     last write. In the worst case the pickled version of this object may be     slightly larger than the blocksize.      Returns:       A dictionary with the state of this object Restore state as part of deserialization/unpickling.      Args:       state: the dictionary from a __getstate__ call Write some bytes. Dummy API.      This API is provided because the zipfile module uses it.  It is a     no-op because Google Storage *requires* that all writes except for     the final one are multiples on 256K bytes aligned on 256K-byte     boundaries. Return the total number of bytes passed to write() so far.      (There is no seek() method.) Flush the buffer and finalize the file.      When this returns the new file is available for reading. Internal API to flush.      This is called only when the total amount of buffered data is at     least self._blocksize, or to flush the final (incomplete) block of     the file with finish=True. Send the block to the storage service and update self._written.