gsutil/gslib/boto_translation.py

# -*- coding: utf-8 -*-
# Copyright 2013 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.
"""XML/boto gsutil Cloud API implementation for GCS and Amazon S3."""

from __future__ import absolute_import

import base64
import binascii
import datetime
import errno
import httplib
import json
import multiprocessing
import os
import pickle
import random
import re
import socket
import tempfile
import textwrap
import threading
import time
import xml
from xml.dom.minidom import parseString as XmlParseString
from xml.sax import _exceptions as SaxExceptions

import boto
from boto import handler
from boto.exception import ResumableDownloadException as BotoResumableDownloadException
from boto.exception import ResumableTransferDisposition
from boto.gs.cors import Cors
from boto.gs.lifecycle import LifecycleConfig
from boto.s3.cors import CORSConfiguration as S3Cors
from boto.s3.deletemarker import DeleteMarker
from boto.s3.lifecycle import Lifecycle as S3Lifecycle
from boto.s3.prefix import Prefix

from gslib.boto_resumable_upload import BotoResumableUpload
from gslib.cloud_api import AccessDeniedException
from gslib.cloud_api import ArgumentException
from gslib.cloud_api import BadRequestException
from gslib.cloud_api import CloudApi
from gslib.cloud_api import NotEmptyException
from gslib.cloud_api import NotFoundException
from gslib.cloud_api import PreconditionException
from gslib.cloud_api import ResumableDownloadException
from gslib.cloud_api import ResumableUploadAbortException
from gslib.cloud_api import ResumableUploadException
from gslib.cloud_api import ResumableUploadStartOverException
from gslib.cloud_api import ServiceException
from gslib.cloud_api_helper import ValidateDstObjectMetadata
# Imported for boto AuthHandler purposes.
import gslib.devshell_auth_plugin  # pylint: disable=unused-import
from gslib.exception import CommandException
from gslib.exception import InvalidUrlError
from gslib.hashing_helper import Base64EncodeHash
from gslib.hashing_helper import Base64ToHexHash
from gslib.project_id import GOOG_PROJ_ID_HDR
from gslib.project_id import PopulateProjectId
from gslib.storage_url import StorageUrlFromString
from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
from gslib.translation_helper import AclTranslation
from gslib.translation_helper import AddS3MarkerAclToObjectMetadata
from gslib.translation_helper import CorsTranslation
from gslib.translation_helper import CreateBucketNotFoundException
from gslib.translation_helper import CreateNotFoundExceptionForObjectWrite
from gslib.translation_helper import CreateObjectNotFoundException
from gslib.translation_helper import DEFAULT_CONTENT_TYPE
from gslib.translation_helper import EncodeStringAsLong
from gslib.translation_helper import GenerationFromUrlAndString
from gslib.translation_helper import HeadersFromObjectMetadata
from gslib.translation_helper import LifecycleTranslation
from gslib.translation_helper import REMOVE_CORS_CONFIG
from gslib.translation_helper import S3MarkerAclFromObjectMetadata
from gslib.util import ConfigureNoOpAuthIfNeeded
from gslib.util import DEFAULT_FILE_BUFFER_SIZE
from gslib.util import GetMaxRetryDelay
from gslib.util import GetNumRetries
from gslib.util import S3_DELETE_MARKER_GUID
from gslib.util import TWO_MIB
from gslib.util import UnaryDictToXml
from gslib.util import UTF8
from gslib.util import XML_PROGRESS_CALLBACKS

TRANSLATABLE_BOTO_EXCEPTIONS = (boto.exception.BotoServerError,
                                boto.exception.InvalidUriError,
                                boto.exception.ResumableDownloadException,
                                boto.exception.ResumableUploadException,
                                boto.exception.StorageCreateError,
                                boto.exception.StorageResponseError)

# pylint: disable=global-at-module-level
global boto_auth_initialized, boto_auth_initialized_lock
# If multiprocessing is available, these will be overridden to process-safe
# variables in InitializeMultiprocessingVariables.
boto_auth_initialized_lock = threading.Lock()
boto_auth_initialized = False

NON_EXISTENT_OBJECT_REGEX = re.compile(r'.*non-\s*existent\s*object',
                                       flags=re.DOTALL)
# Determines whether an etag is a valid MD5.
MD5_REGEX = re.compile(r'^"*[a-fA-F0-9]{32}"*$')


def InitializeMultiprocessingVariables():  # pylint: disable=invalid-name
  """Perform necessary initialization for multiprocessing.

    See gslib.command.InitializeMultiprocessingVariables for an explanation
    of why this is necessary.
  """
  # pylint: disable=global-variable-undefined
  global boto_auth_initialized, boto_auth_initialized_lock
  boto_auth_initialized_lock = gslib.util.CreateLock()
  boto_auth_initialized = multiprocessing.Value('i', 0)


class DownloadProxyCallbackHandler(object):
  """Intermediary callback to keep track of the number of bytes downloaded."""

  def __init__(self, start_byte, callback):
    self._start_byte = start_byte
    self._callback = callback

  def call(self, bytes_downloaded, total_size):
    """Saves necessary data and then calls the given Cloud API callback.

    Args:
      bytes_downloaded: Number of bytes processed so far.
      total_size: Total size of the ongoing operation.
    """
    if self._callback:
      self._callback(self._start_byte + bytes_downloaded, total_size)


class BotoTranslation(CloudApi):
  """Boto-based XML translation implementation of gsutil Cloud API.

  This class takes gsutil Cloud API objects, translates them to XML service
  calls, and translates the results back into gsutil Cloud API objects for
  use by the caller.
  """

  def __init__(self, bucket_storage_uri_class, logger, provider=None,
               credentials=None, debug=0, trace_token=None):
    """Performs necessary setup for interacting with the cloud storage provider.

    Args:
      bucket_storage_uri_class: boto storage_uri class, used by APIs that
                                provide boto translation or mocking.
      logger: logging.logger for outputting log messages.
      provider: Provider prefix describing cloud storage provider to connect to.
                'gs' and 's3' are supported. Function implementations ignore
                the provider argument and use this one instead.
      credentials: Unused.
      debug: Debug level for the API implementation (0..3).
      trace_token: Unused in this subclass.
    """
    super(BotoTranslation, self).__init__(bucket_storage_uri_class, logger,
                                          provider=provider, debug=debug)
    _ = credentials
    # pylint: disable=global-variable-undefined, global-variable-not-assigned
    global boto_auth_initialized, boto_auth_initialized_lock
    with boto_auth_initialized_lock:
      ConfigureNoOpAuthIfNeeded()
      if isinstance(boto_auth_initialized, bool):
        boto_auth_initialized = True
      else:
        boto_auth_initialized.value = 1
    self.api_version = boto.config.get_value(
        'GSUtil', 'default_api_version', '1')

  def GetBucket(self, bucket_name, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    bucket_uri = self._StorageUriForBucket(bucket_name)
    headers = {}
    self._AddApiVersionToHeaders(headers)
    try:
      return self._BotoBucketToBucket(bucket_uri.get_bucket(validate=True,
                                                            headers=headers),
                                      fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def ListBuckets(self, project_id=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    get_fields = self._ListToGetFields(list_fields=fields)
    headers = {}
    self._AddApiVersionToHeaders(headers)
    if self.provider == 'gs':
      headers[GOOG_PROJ_ID_HDR] = PopulateProjectId(project_id)
    try:
      provider_uri = boto.storage_uri(
          '%s://' % self.provider,
          suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)

      buckets_iter = provider_uri.get_all_buckets(headers=headers)
      for bucket in buckets_iter:
        if self.provider == 's3' and bucket.name.lower() != bucket.name:
          # S3 listings can return buckets with upper-case names, but boto
          # can't successfully call them.
          continue
        yield self._BotoBucketToBucket(bucket, fields=get_fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

  def PatchBucket(self, bucket_name, metadata, canned_acl=None,
                  canned_def_acl=None, preconditions=None, provider=None,
                  fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    bucket_uri = self._StorageUriForBucket(bucket_name)
    headers = {}
    self._AddApiVersionToHeaders(headers)
    try:
      self._AddPreconditionsToHeaders(preconditions, headers)
      if metadata.acl:
        boto_acl = AclTranslation.BotoAclFromMessage(metadata.acl)
        bucket_uri.set_xml_acl(boto_acl.to_xml(), headers=headers)
      if canned_acl:
        canned_acls = bucket_uri.canned_acls()
        if canned_acl not in canned_acls:
          raise CommandException('Invalid canned ACL "%s".' % canned_acl)
        bucket_uri.set_acl(canned_acl, bucket_uri.object_name)
      if canned_def_acl:
        canned_acls = bucket_uri.canned_acls()
        if canned_def_acl not in canned_acls:
          raise CommandException('Invalid canned ACL "%s".' % canned_def_acl)
        bucket_uri.set_def_acl(canned_def_acl, bucket_uri.object_name)
      if metadata.cors:
        if metadata.cors == REMOVE_CORS_CONFIG:
          metadata.cors = []
        boto_cors = CorsTranslation.BotoCorsFromMessage(metadata.cors)
        bucket_uri.set_cors(boto_cors, False)
      if metadata.defaultObjectAcl:
        boto_acl = AclTranslation.BotoAclFromMessage(
            metadata.defaultObjectAcl)
        bucket_uri.set_def_xml_acl(boto_acl.to_xml(), headers=headers)
      if metadata.lifecycle:
        boto_lifecycle = LifecycleTranslation.BotoLifecycleFromMessage(
            metadata.lifecycle)
        bucket_uri.configure_lifecycle(boto_lifecycle, False)
      if metadata.logging:
        if self.provider == 'gs':
          headers[GOOG_PROJ_ID_HDR] = PopulateProjectId(None)
        if metadata.logging.logBucket and metadata.logging.logObjectPrefix:
          bucket_uri.enable_logging(metadata.logging.logBucket,
                                    metadata.logging.logObjectPrefix,
                                    False, headers)
        else:  # Logging field is present and empty.  Disable logging.
          bucket_uri.disable_logging(False, headers)
      if metadata.versioning:
        bucket_uri.configure_versioning(metadata.versioning.enabled,
                                        headers=headers)
      if metadata.website:
        main_page_suffix = metadata.website.mainPageSuffix
        error_page = metadata.website.notFoundPage
        bucket_uri.set_website_config(main_page_suffix, error_page)
      return self.GetBucket(bucket_name, fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def CreateBucket(self, bucket_name, project_id=None, metadata=None,
                   provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    bucket_uri = self._StorageUriForBucket(bucket_name)
    location = ''
    if metadata and metadata.location:
      location = metadata.location
    # Pass storage_class param only if this is a GCS bucket. (In S3 the
    # storage class is specified on the key object.)
    headers = {}
    if bucket_uri.scheme == 'gs':
      self._AddApiVersionToHeaders(headers)
      headers[GOOG_PROJ_ID_HDR] = PopulateProjectId(project_id)
      storage_class = ''
      if metadata and metadata.storageClass:
        storage_class = metadata.storageClass
      try:
        bucket_uri.create_bucket(headers=headers, location=location,
                                 storage_class=storage_class)
      except TRANSLATABLE_BOTO_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)
    else:
      try:
        bucket_uri.create_bucket(headers=headers, location=location)
      except TRANSLATABLE_BOTO_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)
    return self.GetBucket(bucket_name, fields=fields)

  def DeleteBucket(self, bucket_name, preconditions=None, provider=None):
    """See CloudApi class for function doc strings."""
    _ = provider, preconditions
    bucket_uri = self._StorageUriForBucket(bucket_name)
    headers = {}
    self._AddApiVersionToHeaders(headers)
    try:
      bucket_uri.delete_bucket(headers=headers)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      translated_exception = self._TranslateBotoException(
          e, bucket_name=bucket_name)
      if (translated_exception and
          'BucketNotEmpty' in translated_exception.reason):
        try:
          if bucket_uri.get_versioning_config():
            if self.provider == 's3':
              raise NotEmptyException(
                  'VersionedBucketNotEmpty (%s). Currently, gsutil does not '
                  'support listing or removing S3 DeleteMarkers, so you may '
                  'need to delete these using another tool to successfully '
                  'delete this bucket.' % bucket_name, status=e.status)
            raise NotEmptyException(
                'VersionedBucketNotEmpty (%s)' % bucket_name, status=e.status)
          else:
            raise NotEmptyException('BucketNotEmpty (%s)' % bucket_name,
                                    status=e.status)
        except TRANSLATABLE_BOTO_EXCEPTIONS, e2:
          self._TranslateExceptionAndRaise(e2, bucket_name=bucket_name)
      elif translated_exception and translated_exception.status == 404:
        raise NotFoundException('Bucket %s does not exist.' % bucket_name)
      else:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def ListObjects(self, bucket_name, prefix=None, delimiter=None,
                  all_versions=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    get_fields = self._ListToGetFields(list_fields=fields)
    bucket_uri = self._StorageUriForBucket(bucket_name)
    headers = {}
    yield_prefixes = fields is None or 'prefixes' in fields
    yield_objects = fields is None or any(
        field.startswith('items/') for field in fields)
    self._AddApiVersionToHeaders(headers)
    try:
      objects_iter = bucket_uri.list_bucket(prefix=prefix or '',
                                            delimiter=delimiter or '',
                                            all_versions=all_versions,
                                            headers=headers)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

    try:
      for key in objects_iter:
        if yield_prefixes and isinstance(key, Prefix):
          yield CloudApi.CsObjectOrPrefix(key.name,
                                          CloudApi.CsObjectOrPrefixType.PREFIX)
        elif yield_objects:
          key_to_convert = key

          # Listed keys are populated with these fields during bucket listing.
          key_http_fields = set(['bucket', 'etag', 'name', 'updated',
                                 'generation', 'metageneration', 'size'])

          # When fields == None, the caller is requesting all possible fields.
          # If the caller requested any fields that are not populated by bucket
          # listing, we'll need to make a separate HTTP call for each object to
          # get its metadata and populate the remaining fields with the result.
          if not get_fields or (get_fields and not
                                get_fields.issubset(key_http_fields)):

            generation = None
            if getattr(key, 'generation', None):
              generation = key.generation
            if getattr(key, 'version_id', None):
              generation = key.version_id
            key_to_convert = self._GetBotoKey(bucket_name, key.name,
                                              generation=generation)
          return_object = self._BotoKeyToObject(key_to_convert,
                                                fields=get_fields)

          yield CloudApi.CsObjectOrPrefix(return_object,
                                          CloudApi.CsObjectOrPrefixType.OBJECT)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def GetObjectMetadata(self, bucket_name, object_name, generation=None,
                        provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    try:
      return self._BotoKeyToObject(self._GetBotoKey(bucket_name, object_name,
                                                    generation=generation),
                                   fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _CurryDigester(self, digester_object):
    """Curries a digester object into a form consumable by boto.

    Key instantiates its own digesters by calling hash_algs[alg]() [note there
    are no arguments to this function].  So in order to pass in our caught-up
    digesters during a resumable download, we need to pass the digester
    object but don't get to look it up based on the algorithm name.  Here we
    use a lambda to make lookup implicit.

    Args:
      digester_object: Input object to be returned by the created function.

    Returns:
      A function which when called will return the input object.
    """
    return lambda: digester_object

  def GetObjectMedia(
      self, bucket_name, object_name, download_stream, provider=None,
      generation=None, object_size=None,
      download_strategy=CloudApi.DownloadStrategy.ONE_SHOT,
      start_byte=0, end_byte=None, progress_callback=None,
      serialization_data=None, digesters=None):
    """See CloudApi class for function doc strings."""
    # This implementation will get the object metadata first if we don't pass it
    # in via serialization_data.
    headers = {}
    self._AddApiVersionToHeaders(headers)
    if 'accept-encoding' not in headers:
      headers['accept-encoding'] = 'gzip'
    if end_byte is not None:
      headers['range'] = 'bytes=%s-%s' % (start_byte, end_byte)
    elif start_byte > 0:
      headers['range'] = 'bytes=%s-' % (start_byte)
    elif start_byte < 0:
      headers['range'] = 'bytes=%s' % (start_byte)

    # Since in most cases we already made a call to get the object metadata,
    # here we avoid an extra HTTP call by unpickling the key.  This is coupled
    # with the implementation in _BotoKeyToObject.
    if serialization_data:
      serialization_dict = json.loads(serialization_data)
      key = pickle.loads(binascii.a2b_base64(serialization_dict['url']))
    else:
      key = self._GetBotoKey(bucket_name, object_name, generation=generation)

    if digesters and self.provider == 'gs':
      hash_algs = {}
      for alg in digesters:
        hash_algs[alg] = self._CurryDigester(digesters[alg])
    else:
      hash_algs = {}

    total_size = object_size or 0
    if serialization_data:
      total_size = json.loads(serialization_data)['total_size']

    if total_size:
      num_progress_callbacks = max(int(total_size) / TWO_MIB,
                                   XML_PROGRESS_CALLBACKS)
    else:
      num_progress_callbacks = XML_PROGRESS_CALLBACKS

    try:
      if download_strategy is CloudApi.DownloadStrategy.RESUMABLE:
        self._PerformResumableDownload(
            download_stream, start_byte, end_byte, key,
            headers=headers, callback=progress_callback,
            num_callbacks=num_progress_callbacks, hash_algs=hash_algs)
      elif download_strategy is CloudApi.DownloadStrategy.ONE_SHOT:
        self._PerformSimpleDownload(
            download_stream, key, progress_callback=progress_callback,
            num_progress_callbacks=num_progress_callbacks, headers=headers,
            hash_algs=hash_algs)
      else:
        raise ArgumentException('Unsupported DownloadStrategy: %s' %
                                download_strategy)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

    if self.provider == 's3':
      if digesters:

        class HashToDigester(object):
          """Wrapper class to expose hash digests.

          boto creates its own digesters in s3's get_file, returning on-the-fly
          hashes only by way of key.local_hashes.  To propagate the digest back
          to the caller, this stub class implements the digest() function.
          """

          def __init__(self, hash_val):
            self.hash_val = hash_val

          def digest(self):  # pylint: disable=invalid-name
            return self.hash_val

        for alg_name in digesters:
          if ((download_strategy == CloudApi.DownloadStrategy.RESUMABLE and
               start_byte != 0) or
              not ((getattr(key, 'local_hashes', None) and
                    alg_name in key.local_hashes))):
            # For resumable downloads, boto does not provide a mechanism to
            # catch up the hash in the case of a partially complete download.
            # In this case or in the case where no digest was successfully
            # calculated, set the digester to None, which indicates that we'll
            # need to manually calculate the hash from the local file once it
            # is complete.
            digesters[alg_name] = None
          else:
            # Use the on-the-fly hash.
            digesters[alg_name] = HashToDigester(key.local_hashes[alg_name])

  def _PerformSimpleDownload(self, download_stream, key, progress_callback=None,
                             num_progress_callbacks=XML_PROGRESS_CALLBACKS,
                             headers=None, hash_algs=None):
    if not headers:
      headers = {}
      self._AddApiVersionToHeaders(headers)
    try:
      key.get_contents_to_file(download_stream, cb=progress_callback,
                               num_cb=num_progress_callbacks, headers=headers,
                               hash_algs=hash_algs)
    except TypeError:  # s3 and mocks do not support hash_algs
      key.get_contents_to_file(download_stream, cb=progress_callback,
                               num_cb=num_progress_callbacks, headers=headers)

  def _PerformResumableDownload(self, fp, start_byte, end_byte, key,
                                headers=None, callback=None,
                                num_callbacks=XML_PROGRESS_CALLBACKS,
                                hash_algs=None):
    """Downloads bytes from key to fp, resuming as needed.

    Args:
      fp: File pointer into which data should be downloaded.
      start_byte: Start byte of the download.
      end_byte: End byte of the download.
      key: Key object from which data is to be downloaded
      headers: Headers to send when retrieving the file
      callback: (optional) a callback function that will be called to report
           progress on the download.  The callback should accept two integer
           parameters.  The first integer represents the number of
           bytes that have been successfully transmitted from the service.  The
           second represents the total number of bytes that need to be
           transmitted.
      num_callbacks: (optional) If a callback is specified with the callback
           parameter, this determines the granularity of the callback
           by defining the maximum number of times the callback will be
           called during the file transfer.
      hash_algs: Dict of hash algorithms to apply to downloaded bytes.

    Raises:
      ResumableDownloadException on error.
    """
    if not headers:
      headers = {}
      self._AddApiVersionToHeaders(headers)

    retryable_exceptions = (httplib.HTTPException, IOError, socket.error,
                            socket.gaierror)

    debug = key.bucket.connection.debug

    num_retries = GetNumRetries()
    progress_less_iterations = 0
    last_progress_byte = start_byte

    while True:  # Retry as long as we're making progress.
      try:
        cb_handler = DownloadProxyCallbackHandler(start_byte, callback)
        headers = headers.copy()
        headers['Range'] = 'bytes=%d-%d' % (start_byte, end_byte)

        # Disable AWSAuthConnection-level retry behavior, since that would
        # cause downloads to restart from scratch.
        try:
          key.get_file(fp, headers, cb_handler.call, num_callbacks,
                       override_num_retries=0, hash_algs=hash_algs)
        except TypeError:
          key.get_file(fp, headers, cb_handler.call, num_callbacks,
                       override_num_retries=0)
        fp.flush()
        # Download succeeded.
        return
      except retryable_exceptions, e:
        if debug >= 1:
          self.logger.info('Caught exception (%s)', repr(e))
        if isinstance(e, IOError) and e.errno == errno.EPIPE:
          # Broken pipe error causes httplib to immediately
          # close the socket (http://bugs.python.org/issue5542),
          # so we need to close and reopen the key before resuming
          # the download.
          if self.provider == 's3':
            key.get_file(fp, headers, cb_handler.call, num_callbacks,
                         override_num_retries=0)
          else:  # self.provider == 'gs'
            key.get_file(fp, headers, cb_handler.call, num_callbacks,
                         override_num_retries=0, hash_algs=hash_algs)
      except BotoResumableDownloadException, e:
        if (e.disposition ==
            ResumableTransferDisposition.ABORT_CUR_PROCESS):
          raise ResumableDownloadException(e.message)
        else:
          if debug >= 1:
            self.logger.info('Caught ResumableDownloadException (%s) - will '
                             'retry', e.message)

      # At this point we had a re-tryable failure; see if made progress.
      start_byte = fp.tell()
      if start_byte > last_progress_byte:
        last_progress_byte = start_byte
        progress_less_iterations = 0
      else:
        progress_less_iterations += 1

      if progress_less_iterations > num_retries:
        # Don't retry any longer in the current process.
        raise ResumableDownloadException(
            'Too many resumable download attempts failed without '
            'progress. You might try this download again later')

      # Close the key, in case a previous download died partway
      # through and left data in the underlying key HTTP buffer.
      # Do this within a try/except block in case the connection is
      # closed (since key.close() attempts to do a final read, in which
      # case this read attempt would get an IncompleteRead exception,
      # which we can safely ignore).
      try:
        key.close()
      except httplib.IncompleteRead:
        pass

      sleep_time_secs = min(random.random() * (2 ** progress_less_iterations),
                            GetMaxRetryDelay())
      if debug >= 1:
        self.logger.info(
            'Got retryable failure (%d progress-less in a row).\nSleeping %d '
            'seconds before re-trying', progress_less_iterations,
            sleep_time_secs)
      time.sleep(sleep_time_secs)

  def PatchObjectMetadata(self, bucket_name, object_name, metadata,
                          canned_acl=None, generation=None, preconditions=None,
                          provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    object_uri = self._StorageUriForObject(bucket_name, object_name,
                                           generation=generation)

    headers = {}
    self._AddApiVersionToHeaders(headers)
    meta_headers = HeadersFromObjectMetadata(metadata, self.provider)

    metadata_plus = {}
    metadata_minus = set()
    metadata_changed = False
    for k, v in meta_headers.iteritems():
      metadata_changed = True
      if v is None:
        metadata_minus.add(k)
      else:
        metadata_plus[k] = v

    self._AddPreconditionsToHeaders(preconditions, headers)

    if metadata_changed:
      try:
        object_uri.set_metadata(metadata_plus, metadata_minus, False,
                                headers=headers)
      except TRANSLATABLE_BOTO_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                         object_name=object_name,
                                         generation=generation)

    if metadata.acl:
      boto_acl = AclTranslation.BotoAclFromMessage(metadata.acl)
      try:
        object_uri.set_xml_acl(boto_acl.to_xml(), key_name=object_name)
      except TRANSLATABLE_BOTO_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                         object_name=object_name,
                                         generation=generation)
    if canned_acl:
      canned_acls = object_uri.canned_acls()
      if canned_acl not in canned_acls:
        raise CommandException('Invalid canned ACL "%s".' % canned_acl)
      object_uri.set_acl(canned_acl, object_uri.object_name)

    return self.GetObjectMetadata(bucket_name, object_name,
                                  generation=generation, fields=fields)

  def _PerformSimpleUpload(self, dst_uri, upload_stream, md5=None,
                           canned_acl=None, progress_callback=None,
                           headers=None):
    dst_uri.set_contents_from_file(upload_stream, md5=md5, policy=canned_acl,
                                   cb=progress_callback, headers=headers)

  def _PerformStreamingUpload(self, dst_uri, upload_stream, canned_acl=None,
                              progress_callback=None, headers=None):
    if dst_uri.get_provider().supports_chunked_transfer():
      dst_uri.set_contents_from_stream(upload_stream, policy=canned_acl,
                                       cb=progress_callback, headers=headers)
    else:
      # Provider doesn't support chunked transfer, so copy to a temporary
      # file.
      (temp_fh, temp_path) = tempfile.mkstemp()
      try:
        with open(temp_path, 'wb') as out_fp:
          stream_bytes = upload_stream.read(DEFAULT_FILE_BUFFER_SIZE)
          while stream_bytes:
            out_fp.write(stream_bytes)
            stream_bytes = upload_stream.read(DEFAULT_FILE_BUFFER_SIZE)
        with open(temp_path, 'rb') as in_fp:
          dst_uri.set_contents_from_file(in_fp, policy=canned_acl,
                                         headers=headers)
      finally:
        os.close(temp_fh)
        os.unlink(temp_path)

  def _PerformResumableUpload(self, key, upload_stream, upload_size,
                              tracker_callback, canned_acl=None,
                              serialization_data=None, progress_callback=None,
                              headers=None):
    resumable_upload = BotoResumableUpload(
        tracker_callback, self.logger, resume_url=serialization_data)
    resumable_upload.SendFile(key, upload_stream, upload_size,
                              canned_acl=canned_acl, cb=progress_callback,
                              headers=headers)

  def _UploadSetup(self, object_metadata, preconditions=None):
    """Shared upload implementation.

    Args:
      object_metadata: Object metadata describing destination object.
      preconditions: Optional gsutil Cloud API preconditions.

    Returns:
      Headers dictionary, StorageUri for upload (based on inputs)
    """
    ValidateDstObjectMetadata(object_metadata)

    headers = HeadersFromObjectMetadata(object_metadata, self.provider)
    self._AddApiVersionToHeaders(headers)

    if object_metadata.crc32c:
      if 'x-goog-hash' in headers:
        headers['x-goog-hash'] += (
            ',crc32c=%s' % object_metadata.crc32c.rstrip('\n'))
      else:
        headers['x-goog-hash'] = (
            'crc32c=%s' % object_metadata.crc32c.rstrip('\n'))
    if object_metadata.md5Hash:
      if 'x-goog-hash' in headers:
        headers['x-goog-hash'] += (
            ',md5=%s' % object_metadata.md5Hash.rstrip('\n'))
      else:
        headers['x-goog-hash'] = (
            'md5=%s' % object_metadata.md5Hash.rstrip('\n'))

    if 'content-type' in headers and not headers['content-type']:
      headers['content-type'] = 'application/octet-stream'

    self._AddPreconditionsToHeaders(preconditions, headers)

    dst_uri = self._StorageUriForObject(object_metadata.bucket,
                                        object_metadata.name)
    return headers, dst_uri

  def _HandleSuccessfulUpload(self, dst_uri, object_metadata, fields=None):
    """Set ACLs on an uploaded object and return its metadata.

    Args:
      dst_uri: Generation-specific StorageUri describing the object.
      object_metadata: Metadata for the object, including an ACL if applicable.
      fields: If present, return only these Object metadata fields.

    Returns:
      gsutil Cloud API Object metadata.

    Raises:
      CommandException if the object was overwritten / deleted concurrently.
    """
    try:
      # The XML API does not support if-generation-match for GET requests.
      # Therefore, if the object gets overwritten before the ACL and get_key
      # operations, the best we can do is warn that it happened.
      self._SetObjectAcl(object_metadata, dst_uri)
      return self._BotoKeyToObject(dst_uri.get_key(), fields=fields)
    except boto.exception.InvalidUriError as e:
      if e.message and NON_EXISTENT_OBJECT_REGEX.match(e.message.encode(UTF8)):
        raise CommandException('\n'.join(textwrap.wrap(
            'Uploaded object (%s) was deleted or overwritten immediately '
            'after it was uploaded. This can happen if you attempt to upload '
            'to the same object multiple times concurrently.' % dst_uri.uri)))
      else:
        raise

  def _SetObjectAcl(self, object_metadata, dst_uri):
    """Sets the ACL (if present in object_metadata) on an uploaded object."""
    if object_metadata.acl:
      boto_acl = AclTranslation.BotoAclFromMessage(object_metadata.acl)
      dst_uri.set_xml_acl(boto_acl.to_xml())
    elif self.provider == 's3':
      s3_acl = S3MarkerAclFromObjectMetadata(object_metadata)
      if s3_acl:
        dst_uri.set_xml_acl(s3_acl)

  def UploadObjectResumable(
      self, upload_stream, object_metadata, canned_acl=None, preconditions=None,
      provider=None, fields=None, size=None, serialization_data=None,
      tracker_callback=None, progress_callback=None):
    """See CloudApi class for function doc strings."""
    if self.provider == 's3':
      # Resumable uploads are not supported for s3.
      return self.UploadObject(
          upload_stream, object_metadata, canned_acl=canned_acl,
          preconditions=preconditions, fields=fields, size=size)
    headers, dst_uri = self._UploadSetup(object_metadata,
                                         preconditions=preconditions)
    if not tracker_callback:
      raise ArgumentException('No tracker callback function set for '
                              'resumable upload of %s' % dst_uri)
    try:
      self._PerformResumableUpload(dst_uri.new_key(headers=headers),
                                   upload_stream, size, tracker_callback,
                                   canned_acl=canned_acl,
                                   serialization_data=serialization_data,
                                   progress_callback=progress_callback,
                                   headers=headers)
      return self._HandleSuccessfulUpload(dst_uri, object_metadata,
                                          fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      not_found_exception = CreateNotFoundExceptionForObjectWrite(
          self.provider, object_metadata.bucket)
      self._TranslateExceptionAndRaise(e, bucket_name=object_metadata.bucket,
                                       object_name=object_metadata.name,
                                       not_found_exception=not_found_exception)

  def UploadObjectStreaming(self, upload_stream, object_metadata,
                            canned_acl=None, progress_callback=None,
                            preconditions=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    headers, dst_uri = self._UploadSetup(object_metadata,
                                         preconditions=preconditions)

    try:
      self._PerformStreamingUpload(
          dst_uri, upload_stream, canned_acl=canned_acl,
          progress_callback=progress_callback, headers=headers)
      return self._HandleSuccessfulUpload(dst_uri, object_metadata,
                                          fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      not_found_exception = CreateNotFoundExceptionForObjectWrite(
          self.provider, object_metadata.bucket)
      self._TranslateExceptionAndRaise(e, bucket_name=object_metadata.bucket,
                                       object_name=object_metadata.name,
                                       not_found_exception=not_found_exception)

  def UploadObject(self, upload_stream, object_metadata, canned_acl=None,
                   preconditions=None, size=None, progress_callback=None,
                   provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    headers, dst_uri = self._UploadSetup(object_metadata,
                                         preconditions=preconditions)

    try:
      md5 = None
      if object_metadata.md5Hash:
        md5 = []
        # boto expects hex at index 0, base64 at index 1
        md5.append(Base64ToHexHash(object_metadata.md5Hash))
        md5.append(object_metadata.md5Hash.strip('\n"\''))
      self._PerformSimpleUpload(dst_uri, upload_stream, md5=md5,
                                canned_acl=canned_acl,
                                progress_callback=progress_callback,
                                headers=headers)
      return self._HandleSuccessfulUpload(dst_uri, object_metadata,
                                          fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      not_found_exception = CreateNotFoundExceptionForObjectWrite(
          self.provider, object_metadata.bucket)
      self._TranslateExceptionAndRaise(e, bucket_name=object_metadata.bucket,
                                       object_name=object_metadata.name,
                                       not_found_exception=not_found_exception)

  def DeleteObject(self, bucket_name, object_name, preconditions=None,
                   generation=None, provider=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    headers = {}
    self._AddApiVersionToHeaders(headers)
    self._AddPreconditionsToHeaders(preconditions, headers)

    uri = self._StorageUriForObject(bucket_name, object_name,
                                    generation=generation)
    try:
      uri.delete_key(validate=False, headers=headers)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def CopyObject(self, src_obj_metadata, dst_obj_metadata, src_generation=None,
                 canned_acl=None, preconditions=None, progress_callback=None,
                 max_bytes_per_call=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider

    if max_bytes_per_call is not None:
      raise NotImplementedError('XML API does not suport max_bytes_per_call')
    dst_uri = self._StorageUriForObject(dst_obj_metadata.bucket,
                                        dst_obj_metadata.name)

    # Usually it's okay to treat version_id and generation as
    # the same, but in this case the underlying boto call determines the
    # provider based on the presence of one or the other.
    src_version_id = None
    if self.provider == 's3':
      src_version_id = src_generation
      src_generation = None

    headers = HeadersFromObjectMetadata(dst_obj_metadata, self.provider)
    self._AddApiVersionToHeaders(headers)
    self._AddPreconditionsToHeaders(preconditions, headers)

    if canned_acl:
      headers[dst_uri.get_provider().acl_header] = canned_acl

    preserve_acl = True if dst_obj_metadata.acl else False
    if self.provider == 's3':
      s3_acl = S3MarkerAclFromObjectMetadata(dst_obj_metadata)
      if s3_acl:
        preserve_acl = True

    try:
      new_key = dst_uri.copy_key(
          src_obj_metadata.bucket, src_obj_metadata.name,
          preserve_acl=preserve_acl, headers=headers,
          src_version_id=src_version_id, src_generation=src_generation)

      return self._BotoKeyToObject(new_key, fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      not_found_exception = CreateNotFoundExceptionForObjectWrite(
          self.provider, dst_obj_metadata.bucket, src_provider=self.provider,
          src_bucket_name=src_obj_metadata.bucket,
          src_object_name=src_obj_metadata.name, src_generation=src_generation)
      self._TranslateExceptionAndRaise(e, bucket_name=dst_obj_metadata.bucket,
                                       object_name=dst_obj_metadata.name,
                                       not_found_exception=not_found_exception)

  def ComposeObject(self, src_objs_metadata, dst_obj_metadata,
                    preconditions=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    _ = provider
    ValidateDstObjectMetadata(dst_obj_metadata)

    dst_obj_name = dst_obj_metadata.name
    dst_obj_metadata.name = None
    dst_bucket_name = dst_obj_metadata.bucket
    dst_obj_metadata.bucket = None
    headers = HeadersFromObjectMetadata(dst_obj_metadata, self.provider)
    if not dst_obj_metadata.contentType:
      dst_obj_metadata.contentType = DEFAULT_CONTENT_TYPE
      headers['content-type'] = dst_obj_metadata.contentType
    self._AddApiVersionToHeaders(headers)
    self._AddPreconditionsToHeaders(preconditions, headers)

    dst_uri = self._StorageUriForObject(dst_bucket_name, dst_obj_name)

    src_components = []
    for src_obj in src_objs_metadata:
      src_uri = self._StorageUriForObject(dst_bucket_name, src_obj.name,
                                          generation=src_obj.generation)
      src_components.append(src_uri)

    try:
      dst_uri.compose(src_components, headers=headers)

      return self.GetObjectMetadata(dst_bucket_name, dst_obj_name,
                                    fields=fields)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, dst_obj_metadata.bucket,
                                       dst_obj_metadata.name)

  def _AddPreconditionsToHeaders(self, preconditions, headers):
    """Adds preconditions (if any) to headers."""
    if preconditions and self.provider == 'gs':
      if preconditions.gen_match is not None:
        headers['x-goog-if-generation-match'] = str(preconditions.gen_match)
      if preconditions.meta_gen_match is not None:
        headers['x-goog-if-metageneration-match'] = str(
            preconditions.meta_gen_match)

  def _AddApiVersionToHeaders(self, headers):
    if self.provider == 'gs':
      headers['x-goog-api-version'] = self.api_version

  def _GetMD5FromETag(self, src_etag):
    """Returns an MD5 from the etag iff the etag is a valid MD5 hash.

    Args:
      src_etag: Object etag for which to return the MD5.

    Returns:
      MD5 in hex string format, or None.
    """
    if src_etag and MD5_REGEX.search(src_etag):
      return src_etag.strip('"\'').lower()

  def _StorageUriForBucket(self, bucket):
    """Returns a boto storage_uri for the given bucket name.

    Args:
      bucket: Bucket name (string).

    Returns:
      Boto storage_uri for the bucket.
    """
    return boto.storage_uri(
        '%s://%s' % (self.provider, bucket),
        suppress_consec_slashes=False,
        bucket_storage_uri_class=self.bucket_storage_uri_class,
        debug=self.debug)

  def _StorageUriForObject(self, bucket, object_name, generation=None):
    """Returns a boto storage_uri for the given object.

    Args:
      bucket: Bucket name (string).
      object_name: Object name (string).
      generation: Generation or version_id of object.  If None, live version
                  of the object is used.

    Returns:
      Boto storage_uri for the object.
    """
    uri_string = '%s://%s/%s' % (self.provider, bucket, object_name)
    if generation:
      uri_string += '#%s' % generation
    return boto.storage_uri(
        uri_string, suppress_consec_slashes=False,
        bucket_storage_uri_class=self.bucket_storage_uri_class,
        debug=self.debug)

  def _GetBotoKey(self, bucket_name, object_name, generation=None):
    """Gets the boto key for an object.

    Args:
      bucket_name: Bucket containing the object.
      object_name: Object name.
      generation: Generation or version of the object to retrieve.

    Returns:
      Boto key for the object.
    """
    object_uri = self._StorageUriForObject(bucket_name, object_name,
                                           generation=generation)
    try:
      key = object_uri.get_key()
      if not key:
        raise CreateObjectNotFoundException('404', self.provider,
                                            bucket_name, object_name,
                                            generation=generation)
      return key
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _ListToGetFields(self, list_fields=None):
    """Removes 'items/' from the input fields and converts it to a set.

    This way field sets requested for ListBucket/ListObject can be used in
    _BotoBucketToBucket and _BotoKeyToObject calls.

    Args:
      list_fields: Iterable fields usable in ListBucket/ListObject calls.

    Returns:
      Set of fields usable in GetBucket/GetObject or
      _BotoBucketToBucket/_BotoKeyToObject calls.
    """
    if list_fields:
      get_fields = set()
      for field in list_fields:
        if field in ['kind', 'nextPageToken', 'prefixes']:
          # These are not actually object / bucket metadata fields.
          # They are fields specific to listing, so we don't consider them.
          continue
        get_fields.add(re.sub(r'items/', '', field))
      return get_fields

  # pylint: disable=too-many-statements
  def _BotoBucketToBucket(self, bucket, fields=None):
    """Constructs an apitools Bucket from a boto bucket.

    Args:
      bucket: Boto bucket.
      fields: If present, construct the apitools Bucket with only this set of
              metadata fields.

    Returns:
      apitools Bucket.
    """
    bucket_uri = self._StorageUriForBucket(bucket.name)

    cloud_api_bucket = apitools_messages.Bucket(name=bucket.name,
                                                id=bucket.name)
    headers = {}
    self._AddApiVersionToHeaders(headers)
    if self.provider == 'gs':
      if not fields or 'storageClass' in fields:
        if hasattr(bucket, 'get_storage_class'):
          cloud_api_bucket.storageClass = bucket.get_storage_class()
      if not fields or 'acl' in fields:
        for acl in AclTranslation.BotoBucketAclToMessage(
            bucket.get_acl(headers=headers)):
          try:
            cloud_api_bucket.acl.append(acl)
          except TRANSLATABLE_BOTO_EXCEPTIONS, e:
            translated_exception = self._TranslateBotoException(
                e, bucket_name=bucket.name)
            if (translated_exception and
                isinstance(translated_exception,
                           AccessDeniedException)):
              # JSON API doesn't differentiate between a blank ACL list
              # and an access denied, so this is intentionally left blank.
              pass
            else:
              self._TranslateExceptionAndRaise(e, bucket_name=bucket.name)
      if not fields or 'cors' in fields:
        try:
          boto_cors = bucket_uri.get_cors()
          cloud_api_bucket.cors = CorsTranslation.BotoCorsToMessage(boto_cors)
        except TRANSLATABLE_BOTO_EXCEPTIONS, e:
          self._TranslateExceptionAndRaise(e, bucket_name=bucket.name)
      if not fields or 'defaultObjectAcl' in fields:
        for acl in AclTranslation.BotoObjectAclToMessage(
            bucket.get_def_acl(headers=headers)):
          try:
            cloud_api_bucket.defaultObjectAcl.append(acl)
          except TRANSLATABLE_BOTO_EXCEPTIONS, e:
            translated_exception = self._TranslateBotoException(
                e, bucket_name=bucket.name)
            if (translated_exception and
                isinstance(translated_exception,
                           AccessDeniedException)):
              # JSON API doesn't differentiate between a blank ACL list
              # and an access denied, so this is intentionally left blank.
              pass
            else:
              self._TranslateExceptionAndRaise(e, bucket_name=bucket.name)
      if not fields or 'lifecycle' in fields:
        try:
          boto_lifecycle = bucket_uri.get_lifecycle_config()
          cloud_api_bucket.lifecycle = (
              LifecycleTranslation.BotoLifecycleToMessage(boto_lifecycle))
        except TRANSLATABLE_BOTO_EXCEPTIONS, e:
          self._TranslateExceptionAndRaise(e, bucket_name=bucket.name)
      if not fields or 'logging' in fields:
        try:
          boto_logging = bucket_uri.get_logging_config()
          if boto_logging and 'Logging' in boto_logging:
            logging_config = boto_logging['Logging']
            log_object_prefix_present = 'LogObjectPrefix' in logging_config
            log_bucket_present = 'LogBucket' in logging_config
            if log_object_prefix_present or log_bucket_present:
              cloud_api_bucket.logging = apitools_messages.Bucket.LoggingValue()
              if log_object_prefix_present:
                cloud_api_bucket.logging.logObjectPrefix = (
                    logging_config['LogObjectPrefix'])
              if log_bucket_present:
                cloud_api_bucket.logging.logBucket = logging_config['LogBucket']
        except TRANSLATABLE_BOTO_EXCEPTIONS, e:
          self._TranslateExceptionAndRaise(e, bucket_name=bucket.name)
      if not fields or 'website' in fields:
        try:
          boto_website = bucket_uri.get_website_config()
          if boto_website and 'WebsiteConfiguration' in boto_website:
            website_config = boto_website['WebsiteConfiguration']
            main_page_suffix_present = 'MainPageSuffix' in website_config
            not_found_page_present = 'NotFoundPage' in website_config
            if main_page_suffix_present or not_found_page_present:
              cloud_api_bucket.website = apitools_messages.Bucket.WebsiteValue()
              if main_page_suffix_present:
                cloud_api_bucket.website.mainPageSuffix = (
                    website_config['MainPageSuffix'])
              if not_found_page_present:
                cloud_api_bucket.website.notFoundPage = (
                    website_config['NotFoundPage'])
        except TRANSLATABLE_BOTO_EXCEPTIONS, e:
          self._TranslateExceptionAndRaise(e, bucket_name=bucket.name)
      if not fields or 'location' in fields:
        cloud_api_bucket.location = bucket_uri.get_location()
    if not fields or 'versioning' in fields:
      versioning = bucket_uri.get_versioning_config(headers=headers)
      if versioning:
        if (self.provider == 's3' and 'Versioning' in versioning and
            versioning['Versioning'] == 'Enabled'):
          cloud_api_bucket.versioning = (
              apitools_messages.Bucket.VersioningValue(enabled=True))
        elif self.provider == 'gs':
          cloud_api_bucket.versioning = (
              apitools_messages.Bucket.VersioningValue(enabled=True))

    # For S3 long bucket listing we do not support CORS, lifecycle, website, and
    # logging translation. The individual commands can be used to get
    # the XML equivalents for S3.
    return cloud_api_bucket

  def _BotoKeyToObject(self, key, fields=None):
    """Constructs an apitools Object from a boto key.

    Args:
      key: Boto key to construct Object from.
      fields: If present, construct the apitools Object with only this set of
              metadata fields.

    Returns:
      apitools Object corresponding to key.
    """
    custom_metadata = None
    if not fields or 'metadata' in fields:
      custom_metadata = self._TranslateBotoKeyCustomMetadata(key)
    cache_control = None
    if not fields or 'cacheControl' in fields:
      cache_control = getattr(key, 'cache_control', None)
    component_count = None
    if not fields or 'componentCount' in fields:
      component_count = getattr(key, 'component_count', None)
    content_disposition = None
    if not fields or 'contentDisposition' in fields:
      content_disposition = getattr(key, 'content_disposition', None)
    # Other fields like updated and ACL depend on the generation
    # of the object, so populate that regardless of whether it was requested.
    generation = self._TranslateBotoKeyGeneration(key)
    metageneration = None
    if not fields or 'metageneration' in fields:
      metageneration = self._TranslateBotoKeyMetageneration(key)
    updated = None
    # Translation code to avoid a dependency on dateutil.
    if not fields or 'updated' in fields:
      updated = self._TranslateBotoKeyTimestamp(key)
    etag = None
    if not fields or 'etag' in fields:
      etag = getattr(key, 'etag', None)
      if etag:
        etag = etag.strip('"\'')
    crc32c = None
    if not fields or 'crc32c' in fields:
      if hasattr(key, 'cloud_hashes') and 'crc32c' in key.cloud_hashes:
        crc32c = base64.encodestring(key.cloud_hashes['crc32c']).rstrip('\n')
    md5_hash = None
    if not fields or 'md5Hash' in fields:
      if hasattr(key, 'cloud_hashes') and 'md5' in key.cloud_hashes:
        md5_hash = base64.encodestring(key.cloud_hashes['md5']).rstrip('\n')
      elif self._GetMD5FromETag(getattr(key, 'etag', None)):
        md5_hash = Base64EncodeHash(self._GetMD5FromETag(key.etag))
      elif self.provider == 's3':
        # S3 etags are MD5s for non-multi-part objects, but multi-part objects
        # (which include all objects >= 5 GB) have a custom checksum
        # implementation that is not currently supported by gsutil.
        self.logger.warn(
            'Non-MD5 etag (%s) present for key %s, data integrity checks are '
            'not possible.', key.etag, key)

    # Serialize the boto key in the media link if it is requested.  This
    # way we can later access the key without adding an HTTP call.
    media_link = None
    if not fields or 'mediaLink' in fields:
      media_link = binascii.b2a_base64(
          pickle.dumps(key, pickle.HIGHEST_PROTOCOL))
    size = None
    if not fields or 'size' in fields:
      size = key.size or 0
    storage_class = None
    if not fields or 'storageClass' in fields:
      storage_class = getattr(key, 'storage_class', None)

    cloud_api_object = apitools_messages.Object(
        bucket=key.bucket.name,
        name=key.name,
        size=size,
        contentEncoding=key.content_encoding,
        contentLanguage=key.content_language,
        contentType=key.content_type,
        cacheControl=cache_control,
        contentDisposition=content_disposition,
        etag=etag,
        crc32c=crc32c,
        md5Hash=md5_hash,
        generation=generation,
        metageneration=metageneration,
        componentCount=component_count,
        updated=updated,
        metadata=custom_metadata,
        mediaLink=media_link,
        storageClass=storage_class)

    # Remaining functions amend cloud_api_object.
    self._TranslateDeleteMarker(key, cloud_api_object)
    if not fields or 'acl' in fields:
      generation_str = GenerationFromUrlAndString(
          StorageUrlFromString(self.provider), generation)
      self._TranslateBotoKeyAcl(key, cloud_api_object,
                                generation=generation_str)

    return cloud_api_object

  def _TranslateBotoKeyCustomMetadata(self, key):
    """Populates an apitools message from custom metadata in the boto key."""
    custom_metadata = None
    if getattr(key, 'metadata', None):
      custom_metadata = apitools_messages.Object.MetadataValue(
          additionalProperties=[])
      for k, v in key.metadata.iteritems():
        if k.lower() == 'content-language':
          # Work around content-language being inserted into custom metadata.
          continue
        custom_metadata.additionalProperties.append(
            apitools_messages.Object.MetadataValue.AdditionalProperty(
                key=k, value=v))
    return custom_metadata

  def _TranslateBotoKeyGeneration(self, key):
    """Returns the generation/version_id number from the boto key if present."""
    generation = None
    if self.provider == 'gs':
      if getattr(key, 'generation', None):
        generation = long(key.generation)
    elif self.provider == 's3':
      if getattr(key, 'version_id', None):
        generation = EncodeStringAsLong(key.version_id)
    return generation

  def _TranslateBotoKeyMetageneration(self, key):
    """Returns the metageneration number from the boto key if present."""
    metageneration = None
    if self.provider == 'gs':
      if getattr(key, 'metageneration', None):
        metageneration = long(key.metageneration)
    return metageneration

  def _TranslateBotoKeyTimestamp(self, key):
    """Parses the timestamp from the boto key into an datetime object.

    This avoids a dependency on dateutil.

    Args:
      key: Boto key to get timestamp from.

    Returns:
      datetime object if string is parsed successfully, None otherwise.
    """
    if key.last_modified:
      if '.' in key.last_modified:
        key_us_timestamp = key.last_modified.rstrip('Z') + '000Z'
      else:
        key_us_timestamp = key.last_modified.rstrip('Z') + '.000000Z'
      fmt = '%Y-%m-%dT%H:%M:%S.%fZ'
      try:
        return datetime.datetime.strptime(key_us_timestamp, fmt)
      except ValueError:
        try:
          # Try alternate format
          fmt = '%a, %d %b %Y %H:%M:%S %Z'
          return datetime.datetime.strptime(key.last_modified, fmt)
        except ValueError:
          # Could not parse the time; leave updated as None.
          return None

  def _TranslateDeleteMarker(self, key, cloud_api_object):
    """Marks deleted objects with a metadata value (for S3 compatibility)."""
    if isinstance(key, DeleteMarker):
      if not cloud_api_object.metadata:
        cloud_api_object.metadata = apitools_messages.Object.MetadataValue()
        cloud_api_object.metadata.additionalProperties = []
      cloud_api_object.metadata.additionalProperties.append(
          apitools_messages.Object.MetadataValue.AdditionalProperty(
              key=S3_DELETE_MARKER_GUID, value=True))

  def _TranslateBotoKeyAcl(self, key, cloud_api_object, generation=None):
    """Updates cloud_api_object with the ACL from the boto key."""
    storage_uri_for_key = self._StorageUriForObject(key.bucket.name, key.name,
                                                    generation=generation)
    headers = {}
    self._AddApiVersionToHeaders(headers)
    try:
      if self.provider == 'gs':
        key_acl = storage_uri_for_key.get_acl(headers=headers)
        # key.get_acl() does not support versioning so we need to use
        # storage_uri to ensure we're getting the versioned ACL.
        for acl in AclTranslation.BotoObjectAclToMessage(key_acl):
          cloud_api_object.acl.append(acl)
      if self.provider == 's3':
        key_acl = key.get_xml_acl(headers=headers)
        # ACLs for s3 are different and we use special markers to represent
        # them in the gsutil Cloud API.
        AddS3MarkerAclToObjectMetadata(cloud_api_object, key_acl)
    except boto.exception.GSResponseError, e:
      if e.status == 403:
        # Consume access denied exceptions to mimic JSON behavior of simply
        # returning None if sufficient permission is not present.  The caller
        # needs to handle the case where the ACL is not populated.
        pass
      else:
        raise

  def _TranslateExceptionAndRaise(self, e, bucket_name=None, object_name=None,
                                  generation=None, not_found_exception=None):
    """Translates a Boto exception and raises the translated or original value.

    Args:
      e: Any Exception.
      bucket_name: Optional bucket name in request that caused the exception.
      object_name: Optional object name in request that caused the exception.
      generation: Optional generation in request that caused the exception.
      not_found_exception: Optional exception to raise in the not-found case.

    Raises:
      Translated CloudApi exception, or the original exception if it was not
      translatable.
    """
    translated_exception = self._TranslateBotoException(
        e, bucket_name=bucket_name, object_name=object_name,
        generation=generation, not_found_exception=not_found_exception)
    if translated_exception:
      raise translated_exception
    else:
      raise

  def _TranslateBotoException(self, e, bucket_name=None, object_name=None,
                              generation=None, not_found_exception=None):
    """Translates boto exceptions into their gsutil Cloud API equivalents.

    Args:
      e: Any exception in TRANSLATABLE_BOTO_EXCEPTIONS.
      bucket_name: Optional bucket name in request that caused the exception.
      object_name: Optional object name in request that caused the exception.
      generation: Optional generation in request that caused the exception.
      not_found_exception: Optional exception to raise in the not-found case.

    Returns:
      CloudStorageApiServiceException for translatable exceptions, None
      otherwise.

    Because we're using isinstance, check for subtypes first.
    """
    if isinstance(e, boto.exception.StorageResponseError):
      if e.status == 400:
        return BadRequestException(e.code, status=e.status, body=e.body)
      elif e.status == 401 or e.status == 403:
        return AccessDeniedException(e.code, status=e.status, body=e.body)
      elif e.status == 404:
        if not_found_exception:
          # The exception is pre-constructed prior to translation; the HTTP
          # status code isn't available at that time.
          setattr(not_found_exception, 'status', e.status)
          return not_found_exception
        elif bucket_name:
          if object_name:
            return CreateObjectNotFoundException(e.status, self.provider,
                                                 bucket_name, object_name,
                                                 generation=generation)
          return CreateBucketNotFoundException(e.status, self.provider,
                                               bucket_name)
        return NotFoundException(e.message, status=e.status, body=e.body)

      elif e.status == 409 and e.code and 'BucketNotEmpty' in e.code:
        return NotEmptyException('BucketNotEmpty (%s)' % bucket_name,
                                 status=e.status, body=e.body)
      elif e.status == 410:
        # 410 errors should always cause us to start over - either the UploadID
        # has expired or there was a server-side problem that requires starting
        # the upload over from scratch.
        return ResumableUploadStartOverException(e.message)
      elif e.status == 412:
        return PreconditionException(e.code, status=e.status, body=e.body)
    if isinstance(e, boto.exception.StorageCreateError):
      return ServiceException('Bucket already exists.', status=e.status,
                              body=e.body)

    if isinstance(e, boto.exception.BotoServerError):
      return ServiceException(e.message, status=e.status, body=e.body)

    if isinstance(e, boto.exception.InvalidUriError):
      # Work around textwrap when searching for this string.
      if e.message and NON_EXISTENT_OBJECT_REGEX.match(e.message.encode(UTF8)):
        return NotFoundException(e.message, status=404)
      return InvalidUrlError(e.message)

    if isinstance(e, boto.exception.ResumableUploadException):
      if e.disposition == boto.exception.ResumableTransferDisposition.ABORT:
        return ResumableUploadAbortException(e.message)
      elif (e.disposition ==
            boto.exception.ResumableTransferDisposition.START_OVER):
        return ResumableUploadStartOverException(e.message)
      else:
        return ResumableUploadException(e.message)

    if isinstance(e, boto.exception.ResumableDownloadException):
      return ResumableDownloadException(e.message)

    return None

  # For function docstrings, see CloudApiDelegator class.
  def XmlPassThroughGetAcl(self, storage_url, def_obj_acl=False):
    """See CloudApiDelegator class for function doc strings."""
    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      if def_obj_acl:
        return uri.get_def_acl()
      else:
        return uri.get_acl()
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

  def XmlPassThroughSetAcl(self, acl_text, storage_url, canned=True,
                           def_obj_acl=False):
    """See CloudApiDelegator class for function doc strings."""
    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      if canned:
        if def_obj_acl:
          canned_acls = uri.canned_acls()
          if acl_text not in canned_acls:
            raise CommandException('Invalid canned ACL "%s".' % acl_text)
          uri.set_def_acl(acl_text, uri.object_name)
        else:
          canned_acls = uri.canned_acls()
          if acl_text not in canned_acls:
            raise CommandException('Invalid canned ACL "%s".' % acl_text)
          uri.set_acl(acl_text, uri.object_name)
      else:
        if def_obj_acl:
          uri.set_def_xml_acl(acl_text, uri.object_name)
        else:
          uri.set_xml_acl(acl_text, uri.object_name)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

  # pylint: disable=catching-non-exception
  def XmlPassThroughSetCors(self, cors_text, storage_url):
    """See CloudApiDelegator class for function doc strings."""
    # Parse XML document and convert into Cors object.
    if storage_url.scheme == 's3':
      cors_obj = S3Cors()
    else:
      cors_obj = Cors()
    h = handler.XmlHandler(cors_obj, None)
    try:
      xml.sax.parseString(cors_text, h)
    except SaxExceptions.SAXParseException, e:
      raise CommandException('Requested CORS is invalid: %s at line %s, '
                             'column %s' % (e.getMessage(), e.getLineNumber(),
                                            e.getColumnNumber()))

    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      uri.set_cors(cors_obj, False)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

  def XmlPassThroughGetCors(self, storage_url):
    """See CloudApiDelegator class for function doc strings."""
    uri = boto.storage_uri(
        storage_url.url_string, suppress_consec_slashes=False,
        bucket_storage_uri_class=self.bucket_storage_uri_class,
        debug=self.debug)
    try:
      cors = uri.get_cors(False)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

    parsed_xml = xml.dom.minidom.parseString(cors.to_xml().encode(UTF8))
    # Pretty-print the XML to make it more easily human editable.
    return parsed_xml.toprettyxml(indent='    ')

  def XmlPassThroughGetLifecycle(self, storage_url):
    """See CloudApiDelegator class for function doc strings."""
    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      lifecycle = uri.get_lifecycle_config(False)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

    parsed_xml = xml.dom.minidom.parseString(lifecycle.to_xml().encode(UTF8))
    # Pretty-print the XML to make it more easily human editable.
    return parsed_xml.toprettyxml(indent='    ')

  def XmlPassThroughSetLifecycle(self, lifecycle_text, storage_url):
    """See CloudApiDelegator class for function doc strings."""
    # Parse XML document and convert into lifecycle object.
    if storage_url.scheme == 's3':
      lifecycle_obj = S3Lifecycle()
    else:
      lifecycle_obj = LifecycleConfig()
    h = handler.XmlHandler(lifecycle_obj, None)
    try:
      xml.sax.parseString(lifecycle_text, h)
    except SaxExceptions.SAXParseException, e:
      raise CommandException(
          'Requested lifecycle config is invalid: %s at line %s, column %s' %
          (e.getMessage(), e.getLineNumber(), e.getColumnNumber()))

    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      uri.configure_lifecycle(lifecycle_obj, False)
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

  def XmlPassThroughGetLogging(self, storage_url):
    """See CloudApiDelegator class for function doc strings."""
    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      logging_config_xml = UnaryDictToXml(uri.get_logging_config())
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

    return XmlParseString(logging_config_xml).toprettyxml()

  def XmlPassThroughGetWebsite(self, storage_url):
    """See CloudApiDelegator class for function doc strings."""
    try:
      uri = boto.storage_uri(
          storage_url.url_string, suppress_consec_slashes=False,
          bucket_storage_uri_class=self.bucket_storage_uri_class,
          debug=self.debug)
      web_config_xml = UnaryDictToXml(uri.get_website_config())
    except TRANSLATABLE_BOTO_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

    return XmlParseString(web_config_xml).toprettyxml()