android-7.0.0_r1/s

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

from __future__ import absolute_import

import httplib
import json
import logging
import os
import socket
import ssl
import time
import traceback

from apitools.base.py import credentials_lib
from apitools.base.py import encoding
from apitools.base.py import exceptions as apitools_exceptions
from apitools.base.py import http_wrapper as apitools_http_wrapper
from apitools.base.py import transfer as apitools_transfer
from apitools.base.py.util import CalculateWaitForRetry

import boto
from boto import config
from gcs_oauth2_boto_plugin import oauth2_helper
import httplib2
import oauth2client
from oauth2client import devshell
from oauth2client import multistore_file

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 Preconditions
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
from gslib.cred_types import CredTypes
from gslib.exception import CommandException
from gslib.gcs_json_media import BytesTransferredContainer
from gslib.gcs_json_media import DownloadCallbackConnectionClassFactory
from gslib.gcs_json_media import HttpWithDownloadStream
from gslib.gcs_json_media import HttpWithNoRetries
from gslib.gcs_json_media import UploadCallbackConnectionClassFactory
from gslib.gcs_json_media import WrapDownloadHttpRequest
from gslib.gcs_json_media import WrapUploadHttpRequest
from gslib.no_op_credentials import NoOpCredentials
from gslib.progress_callback import ProgressCallbackWithBackoff
from gslib.project_id import PopulateProjectId
from gslib.third_party.storage_apitools import storage_v1_client as apitools_client
from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
from gslib.tracker_file import DeleteTrackerFile
from gslib.tracker_file import GetRewriteTrackerFilePath
from gslib.tracker_file import HashRewriteParameters
from gslib.tracker_file import ReadRewriteTrackerFile
from gslib.tracker_file import WriteRewriteTrackerFile
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 PRIVATE_DEFAULT_OBJ_ACL
from gslib.translation_helper import REMOVE_CORS_CONFIG
from gslib.util import GetBotoConfigFileList
from gslib.util import GetCertsFile
from gslib.util import GetCredentialStoreFilename
from gslib.util import GetGceCredentialCacheFilename
from gslib.util import GetJsonResumableChunkSize
from gslib.util import GetMaxRetryDelay
from gslib.util import GetNewHttp
from gslib.util import GetNumRetries
from gslib.util import UTF8


# Implementation supports only 'gs' URLs, so provider is unused.
# pylint: disable=unused-argument

DEFAULT_GCS_JSON_VERSION = 'v1'

NUM_BUCKETS_PER_LIST_PAGE = 1000
NUM_OBJECTS_PER_LIST_PAGE = 1000

TRANSLATABLE_APITOOLS_EXCEPTIONS = (apitools_exceptions.HttpError,
                                    apitools_exceptions.StreamExhausted,
                                    apitools_exceptions.TransferError,
                                    apitools_exceptions.TransferInvalidError)

# TODO: Distribute these exceptions better through apitools and here.
# Right now, apitools is configured not to handle any exceptions on
# uploads/downloads.
# oauth2_client tries to JSON-decode the response, which can result
# in a ValueError if the response was invalid. Until that is fixed in
# oauth2_client, need to handle it here.
HTTP_TRANSFER_EXCEPTIONS = (apitools_exceptions.TransferRetryError,
                            apitools_exceptions.BadStatusCodeError,
                            # TODO: Honor retry-after headers.
                            apitools_exceptions.RetryAfterError,
                            apitools_exceptions.RequestError,
                            httplib.BadStatusLine,
                            httplib.IncompleteRead,
                            httplib.ResponseNotReady,
                            httplib2.ServerNotFoundError,
                            socket.error,
                            socket.gaierror,
                            socket.timeout,
                            ssl.SSLError,
                            ValueError)

_VALIDATE_CERTIFICATES_503_MESSAGE = (
    """Service Unavailable. If you have recently changed
    https_validate_certificates from True to False in your boto configuration
    file, please delete any cached access tokens in your filesystem (at %s)
    and try again.""" % GetCredentialStoreFilename())


class GcsJsonApi(CloudApi):
  """Google Cloud Storage JSON implementation of gsutil Cloud API."""

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

    Args:
      bucket_storage_uri_class: Unused.
      logger: logging.logger for outputting log messages.
      provider: Unused.  This implementation supports only Google Cloud Storage.
      credentials: Credentials to be used for interacting with Google Cloud
                   Storage.
      debug: Debug level for the API implementation (0..3).
      trace_token: Trace token to pass to the API implementation.
    """
    # TODO: Plumb host_header for perfdiag / test_perfdiag.
    # TODO: Add jitter to apitools' http_wrapper retry mechanism.
    super(GcsJsonApi, self).__init__(bucket_storage_uri_class, logger,
                                     provider='gs', debug=debug)
    no_op_credentials = False
    if not credentials:
      loaded_credentials = self._CheckAndGetCredentials(logger)

      if not loaded_credentials:
        loaded_credentials = NoOpCredentials()
        no_op_credentials = True
    else:
      if isinstance(credentials, NoOpCredentials):
        no_op_credentials = True

    self.credentials = credentials or loaded_credentials

    self.certs_file = GetCertsFile()

    self.http = GetNewHttp()

    # Re-use download and upload connections. This class is only called
    # sequentially, but we can share TCP warmed-up connections across calls.
    self.download_http = self._GetNewDownloadHttp()
    self.upload_http = self._GetNewUploadHttp()
    if self.credentials:
      self.authorized_download_http = self.credentials.authorize(
          self.download_http)
      self.authorized_upload_http = self.credentials.authorize(self.upload_http)
    else:
      self.authorized_download_http = self.download_http
      self.authorized_upload_http = self.upload_http

    WrapDownloadHttpRequest(self.authorized_download_http)
    WrapUploadHttpRequest(self.authorized_upload_http)

    self.http_base = 'https://'
    gs_json_host = config.get('Credentials', 'gs_json_host', None)
    self.host_base = gs_json_host or 'www.googleapis.com'

    if not gs_json_host:
      gs_host = config.get('Credentials', 'gs_host', None)
      if gs_host:
        raise ArgumentException(
            'JSON API is selected but gs_json_host is not configured, '
            'while gs_host is configured to %s. Please also configure '
            'gs_json_host and gs_json_port to match your desired endpoint.'
            % gs_host)

    gs_json_port = config.get('Credentials', 'gs_json_port', None)

    if not gs_json_port:
      gs_port = config.get('Credentials', 'gs_port', None)
      if gs_port:
        raise ArgumentException(
            'JSON API is selected but gs_json_port is not configured, '
            'while gs_port is configured to %s. Please also configure '
            'gs_json_host and gs_json_port to match your desired endpoint.'
            % gs_port)
      self.host_port = ''
    else:
      self.host_port = ':' + config.get('Credentials', 'gs_json_port')

    self.api_version = config.get('GSUtil', 'json_api_version',
                                  DEFAULT_GCS_JSON_VERSION)
    self.url_base = (self.http_base + self.host_base + self.host_port + '/' +
                     'storage/' + self.api_version + '/')

    credential_store_key_dict = self._GetCredentialStoreKeyDict(
        self.credentials)

    self.credentials.set_store(
        multistore_file.get_credential_storage_custom_key(
            GetCredentialStoreFilename(), credential_store_key_dict))

    self.num_retries = GetNumRetries()
    self.max_retry_wait = GetMaxRetryDelay()

    log_request = (debug >= 3)
    log_response = (debug >= 3)

    self.global_params = apitools_messages.StandardQueryParameters(
        trace='token:%s' % trace_token) if trace_token else None

    self.api_client = apitools_client.StorageV1(
        url=self.url_base, http=self.http, log_request=log_request,
        log_response=log_response, credentials=self.credentials,
        version=self.api_version, default_global_params=self.global_params)
    self.api_client.max_retry_wait = self.max_retry_wait
    self.api_client.num_retries = self.num_retries

    if no_op_credentials:
      # This API key is not secret and is used to identify gsutil during
      # anonymous requests.
      self.api_client.AddGlobalParam('key',
                                     u'AIzaSyDnacJHrKma0048b13sh8cgxNUwulubmJM')

  def _CheckAndGetCredentials(self, logger):
    configured_cred_types = []
    try:
      if self._HasOauth2UserAccountCreds():
        configured_cred_types.append(CredTypes.OAUTH2_USER_ACCOUNT)
      if self._HasOauth2ServiceAccountCreds():
        configured_cred_types.append(CredTypes.OAUTH2_SERVICE_ACCOUNT)
      if len(configured_cred_types) > 1:
        # We only allow one set of configured credentials. Otherwise, we're
        # choosing one arbitrarily, which can be very confusing to the user
        # (e.g., if only one is authorized to perform some action) and can
        # also mask errors.
        # Because boto merges config files, GCE credentials show up by default
        # for GCE VMs. We don't want to fail when a user creates a boto file
        # with their own credentials, so in this case we'll use the OAuth2
        # user credentials.
        failed_cred_type = None
        raise CommandException(
            ('You have multiple types of configured credentials (%s), which is '
             'not supported. One common way this happens is if you run gsutil '
             'config to create credentials and later run gcloud auth, and '
             'create a second set of credentials. Your boto config path is: '
             '%s. For more help, see "gsutil help creds".')
            % (configured_cred_types, GetBotoConfigFileList()))

      failed_cred_type = CredTypes.OAUTH2_USER_ACCOUNT
      user_creds = self._GetOauth2UserAccountCreds()
      failed_cred_type = CredTypes.OAUTH2_SERVICE_ACCOUNT
      service_account_creds = self._GetOauth2ServiceAccountCreds()
      failed_cred_type = CredTypes.GCE
      gce_creds = self._GetGceCreds()
      failed_cred_type = CredTypes.DEVSHELL
      devshell_creds = self._GetDevshellCreds()
      return user_creds or service_account_creds or gce_creds or devshell_creds
    except:  # pylint: disable=bare-except

      # If we didn't actually try to authenticate because there were multiple
      # types of configured credentials, don't emit this warning.
      if failed_cred_type:
        if os.environ.get('CLOUDSDK_WRAPPER') == '1':
          logger.warn(
              'Your "%s" credentials are invalid. Please run\n'
              '  $ gcloud auth login', failed_cred_type)
        else:
          logger.warn(
              'Your "%s" credentials are invalid. For more help, see '
              '"gsutil help creds", or re-run the gsutil config command (see '
              '"gsutil help config").', failed_cred_type)

      # If there's any set of configured credentials, we'll fail if they're
      # invalid, rather than silently falling back to anonymous config (as
      # boto does). That approach leads to much confusion if users don't
      # realize their credentials are invalid.
      raise

  def _HasOauth2ServiceAccountCreds(self):
    return config.has_option('Credentials', 'gs_service_key_file')

  def _HasOauth2UserAccountCreds(self):
    return config.has_option('Credentials', 'gs_oauth2_refresh_token')

  def _HasGceCreds(self):
    return config.has_option('GoogleCompute', 'service_account')

  def _GetOauth2ServiceAccountCreds(self):
    if self._HasOauth2ServiceAccountCreds():
      return oauth2_helper.OAuth2ClientFromBotoConfig(
          boto.config,
          cred_type=CredTypes.OAUTH2_SERVICE_ACCOUNT).GetCredentials()

  def _GetOauth2UserAccountCreds(self):
    if self._HasOauth2UserAccountCreds():
      return oauth2_helper.OAuth2ClientFromBotoConfig(
          boto.config).GetCredentials()

  def _GetGceCreds(self):
    if self._HasGceCreds():
      try:
        return credentials_lib.GceAssertionCredentials(
            cache_filename=GetGceCredentialCacheFilename())
      except apitools_exceptions.ResourceUnavailableError, e:
        if 'service account' in str(e) and 'does not exist' in str(e):
          return None
        raise

  def _GetDevshellCreds(self):
    try:
      return devshell.DevshellCredentials()
    except devshell.NoDevshellServer:
      return None
    except:
      raise

  def _GetCredentialStoreKeyDict(self, credentials):
    """Disambiguates a credential for caching in a credential store.

    Different credential types have different fields that identify them.
    This function assembles relevant information in a dict and returns it.

    Args:
      credentials: An OAuth2Credentials object.

    Returns:
      Dict of relevant identifiers for credentials.
    """
    # TODO: If scopes ever become available in the credentials themselves,
    # include them in the key dict.
    key_dict = {'api_version': self.api_version}
    # pylint: disable=protected-access
    if isinstance(credentials, devshell.DevshellCredentials):
      key_dict['user_email'] = credentials.user_email
    elif isinstance(credentials,
                    oauth2client.service_account._ServiceAccountCredentials):
      key_dict['_service_account_email'] = credentials._service_account_email
    elif isinstance(credentials,
                    oauth2client.client.SignedJwtAssertionCredentials):
      key_dict['service_account_name'] = credentials.service_account_name
    elif isinstance(credentials, oauth2client.client.OAuth2Credentials):
      if credentials.client_id and credentials.client_id != 'null':
        key_dict['client_id'] = credentials.client_id
      key_dict['refresh_token'] = credentials.refresh_token
    # pylint: enable=protected-access

    return key_dict

  def _GetNewDownloadHttp(self):
    return GetNewHttp(http_class=HttpWithDownloadStream)

  def _GetNewUploadHttp(self):
    """Returns an upload-safe Http object (by disabling httplib2 retries)."""
    return GetNewHttp(http_class=HttpWithNoRetries)

  def GetBucket(self, bucket_name, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsGetRequest
                  .ProjectionValueValuesEnum.full)
    apitools_request = apitools_messages.StorageBucketsGetRequest(
        bucket=bucket_name, projection=projection)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    # Here and in list buckets, we have no way of knowing
    # whether we requested a field and didn't get it because it didn't exist
    # or because we didn't have permission to access it.
    try:
      return self.api_client.buckets.Get(apitools_request,
                                         global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  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."""
    projection = (apitools_messages.StorageBucketsPatchRequest
                  .ProjectionValueValuesEnum.full)
    bucket_metadata = metadata

    if not preconditions:
      preconditions = Preconditions()

    # For blank metadata objects, we need to explicitly call
    # them out to apitools so it will send/erase them.
    apitools_include_fields = []
    for metadata_field in ('metadata', 'lifecycle', 'logging', 'versioning',
                           'website'):
      attr = getattr(bucket_metadata, metadata_field, None)
      if attr and not encoding.MessageToDict(attr):
        setattr(bucket_metadata, metadata_field, None)
        apitools_include_fields.append(metadata_field)

    if bucket_metadata.cors and bucket_metadata.cors == REMOVE_CORS_CONFIG:
      bucket_metadata.cors = []
      apitools_include_fields.append('cors')

    if (bucket_metadata.defaultObjectAcl and
        bucket_metadata.defaultObjectAcl[0] == PRIVATE_DEFAULT_OBJ_ACL):
      bucket_metadata.defaultObjectAcl = []
      apitools_include_fields.append('defaultObjectAcl')

    predefined_acl = None
    if canned_acl:
      # Must null out existing ACLs to apply a canned ACL.
      apitools_include_fields.append('acl')
      predefined_acl = (
          apitools_messages.StorageBucketsPatchRequest.
          PredefinedAclValueValuesEnum(
              self._BucketCannedAclToPredefinedAcl(canned_acl)))

    predefined_def_acl = None
    if canned_def_acl:
      # Must null out existing default object ACLs to apply a canned ACL.
      apitools_include_fields.append('defaultObjectAcl')
      predefined_def_acl = (
          apitools_messages.StorageBucketsPatchRequest.
          PredefinedDefaultObjectAclValueValuesEnum(
              self._ObjectCannedAclToPredefinedAcl(canned_def_acl)))

    apitools_request = apitools_messages.StorageBucketsPatchRequest(
        bucket=bucket_name, bucketResource=bucket_metadata,
        projection=projection,
        ifMetagenerationMatch=preconditions.meta_gen_match,
        predefinedAcl=predefined_acl,
        predefinedDefaultObjectAcl=predefined_def_acl)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    with self.api_client.IncludeFields(apitools_include_fields):
      try:
        return self.api_client.buckets.Patch(apitools_request,
                                             global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e)

  def CreateBucket(self, bucket_name, project_id=None, metadata=None,
                   provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsInsertRequest
                  .ProjectionValueValuesEnum.full)
    if not metadata:
      metadata = apitools_messages.Bucket()
    metadata.name = bucket_name

    if metadata.location:
      metadata.location = metadata.location.upper()
    if metadata.storageClass:
      metadata.storageClass = metadata.storageClass.upper()

    project_id = PopulateProjectId(project_id)

    apitools_request = apitools_messages.StorageBucketsInsertRequest(
        bucket=metadata, project=project_id, projection=projection)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    try:
      return self.api_client.buckets.Insert(apitools_request,
                                            global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def DeleteBucket(self, bucket_name, preconditions=None, provider=None):
    """See CloudApi class for function doc strings."""
    if not preconditions:
      preconditions = Preconditions()

    apitools_request = apitools_messages.StorageBucketsDeleteRequest(
        bucket=bucket_name, ifMetagenerationMatch=preconditions.meta_gen_match)

    try:
      self.api_client.buckets.Delete(apitools_request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      if isinstance(
          self._TranslateApitoolsException(e, bucket_name=bucket_name),
          NotEmptyException):
        # If bucket is not empty, check to see if versioning is enabled and
        # signal that in the exception if it is.
        bucket_metadata = self.GetBucket(bucket_name,
                                         fields=['versioning'])
        if bucket_metadata.versioning and bucket_metadata.versioning.enabled:
          raise NotEmptyException('VersionedBucketNotEmpty',
                                  status=e.status_code)
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def ListBuckets(self, project_id=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsListRequest
                  .ProjectionValueValuesEnum.full)
    project_id = PopulateProjectId(project_id)

    apitools_request = apitools_messages.StorageBucketsListRequest(
        project=project_id, maxResults=NUM_BUCKETS_PER_LIST_PAGE,
        projection=projection)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      if 'nextPageToken' not in fields:
        fields.add('nextPageToken')
      global_params.fields = ','.join(set(fields))
    try:
      bucket_list = self.api_client.buckets.List(apitools_request,
                                                 global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

    for bucket in self._YieldBuckets(bucket_list):
      yield bucket

    while bucket_list.nextPageToken:
      apitools_request = apitools_messages.StorageBucketsListRequest(
          project=project_id, pageToken=bucket_list.nextPageToken,
          maxResults=NUM_BUCKETS_PER_LIST_PAGE, projection=projection)
      try:
        bucket_list = self.api_client.buckets.List(apitools_request,
                                                   global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e)

      for bucket in self._YieldBuckets(bucket_list):
        yield bucket

  def _YieldBuckets(self, bucket_list):
    """Yields buckets from a list returned by apitools."""
    if bucket_list.items:
      for bucket in bucket_list.items:
        yield bucket

  def ListObjects(self, bucket_name, prefix=None, delimiter=None,
                  all_versions=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageObjectsListRequest
                  .ProjectionValueValuesEnum.full)
    apitools_request = apitools_messages.StorageObjectsListRequest(
        bucket=bucket_name, prefix=prefix, delimiter=delimiter,
        versions=all_versions, projection=projection,
        maxResults=NUM_OBJECTS_PER_LIST_PAGE)
    global_params = apitools_messages.StandardQueryParameters()

    if fields:
      fields = set(fields)
      if 'nextPageToken' not in fields:
        fields.add('nextPageToken')
      global_params.fields = ','.join(fields)

    try:
      object_list = self.api_client.objects.List(apitools_request,
                                                 global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

    for object_or_prefix in self._YieldObjectsAndPrefixes(object_list):
      yield object_or_prefix

    while object_list.nextPageToken:
      apitools_request = apitools_messages.StorageObjectsListRequest(
          bucket=bucket_name, prefix=prefix, delimiter=delimiter,
          versions=all_versions, projection=projection,
          pageToken=object_list.nextPageToken,
          maxResults=NUM_OBJECTS_PER_LIST_PAGE)
      try:
        object_list = self.api_client.objects.List(apitools_request,
                                                   global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

      for object_or_prefix in self._YieldObjectsAndPrefixes(object_list):
        yield object_or_prefix

  def _YieldObjectsAndPrefixes(self, object_list):
    # Yield prefixes first so that checking for the presence of a subdirectory
    # is fast.
    if object_list.prefixes:
      for prefix in object_list.prefixes:
        yield CloudApi.CsObjectOrPrefix(prefix,
                                        CloudApi.CsObjectOrPrefixType.PREFIX)
    if object_list.items:
      for cloud_obj in object_list.items:
        yield CloudApi.CsObjectOrPrefix(cloud_obj,
                                        CloudApi.CsObjectOrPrefixType.OBJECT)

  def GetObjectMetadata(self, bucket_name, object_name, generation=None,
                        provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageObjectsGetRequest
                  .ProjectionValueValuesEnum.full)

    if generation:
      generation = long(generation)

    apitools_request = apitools_messages.StorageObjectsGetRequest(
        bucket=bucket_name, object=object_name, projection=projection,
        generation=generation)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    try:
      return self.api_client.objects.Get(apitools_request,
                                         global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  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.
    if generation:
      generation = long(generation)

    # 'outer_total_size' is only used for formatting user output, and is
    # expected to be one higher than the last byte that should be downloaded.
    # TODO: Change DownloadCallbackConnectionClassFactory and progress callbacks
    # to more elegantly handle total size for components of files.
    outer_total_size = object_size
    if end_byte:
      outer_total_size = end_byte + 1
    elif serialization_data:
      outer_total_size = json.loads(serialization_data)['total_size']

    if progress_callback:
      if outer_total_size is None:
        raise ArgumentException('Download size is required when callbacks are '
                                'requested for a download, but no size was '
                                'provided.')
      progress_callback(start_byte, outer_total_size)

    bytes_downloaded_container = BytesTransferredContainer()
    bytes_downloaded_container.bytes_transferred = start_byte

    callback_class_factory = DownloadCallbackConnectionClassFactory(
        bytes_downloaded_container, total_size=outer_total_size,
        progress_callback=progress_callback, digesters=digesters)
    download_http_class = callback_class_factory.GetConnectionClass()

    # Point our download HTTP at our download stream.
    self.download_http.stream = download_stream
    self.download_http.connections = {'https': download_http_class}

    if serialization_data:
      apitools_download = apitools_transfer.Download.FromData(
          download_stream, serialization_data, self.api_client.http,
          num_retries=self.num_retries)
    else:
      apitools_download = apitools_transfer.Download.FromStream(
          download_stream, auto_transfer=False, total_size=object_size,
          num_retries=self.num_retries)

    apitools_download.bytes_http = self.authorized_download_http
    apitools_request = apitools_messages.StorageObjectsGetRequest(
        bucket=bucket_name, object=object_name, generation=generation)

    try:
      if download_strategy == CloudApi.DownloadStrategy.RESUMABLE:
        # Disable retries in apitools. We will handle them explicitly here.
        apitools_download.retry_func = (
            apitools_http_wrapper.RethrowExceptionHandler)
        return self._PerformResumableDownload(
            bucket_name, object_name, download_stream, apitools_request,
            apitools_download, bytes_downloaded_container,
            generation=generation, start_byte=start_byte, end_byte=end_byte,
            serialization_data=serialization_data)
      else:
        return self._PerformDownload(
            bucket_name, object_name, download_stream, apitools_request,
            apitools_download, generation=generation, start_byte=start_byte,
            end_byte=end_byte, serialization_data=serialization_data)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _PerformResumableDownload(
      self, bucket_name, object_name, download_stream, apitools_request,
      apitools_download, bytes_downloaded_container, generation=None,
      start_byte=0, end_byte=None, serialization_data=None):
    retries = 0
    last_progress_byte = start_byte
    while retries <= self.num_retries:
      try:
        return self._PerformDownload(
            bucket_name, object_name, download_stream, apitools_request,
            apitools_download, generation=generation, start_byte=start_byte,
            end_byte=end_byte, serialization_data=serialization_data)
      except HTTP_TRANSFER_EXCEPTIONS, e:
        start_byte = download_stream.tell()
        bytes_downloaded_container.bytes_transferred = start_byte
        if start_byte > last_progress_byte:
          # We've made progress, so allow a fresh set of retries.
          last_progress_byte = start_byte
          retries = 0
        retries += 1
        if retries > self.num_retries:
          raise ResumableDownloadException(
              'Transfer failed after %d retries. Final exception: %s' %
              (self.num_retries, unicode(e).encode(UTF8)))
        time.sleep(CalculateWaitForRetry(retries, max_wait=self.max_retry_wait))
        if self.logger.isEnabledFor(logging.DEBUG):
          self.logger.debug(
              'Retrying download from byte %s after exception: %s. Trace: %s',
              start_byte, unicode(e).encode(UTF8), traceback.format_exc())
        apitools_http_wrapper.RebuildHttpConnections(
            apitools_download.bytes_http)

  def _PerformDownload(
      self, bucket_name, object_name, download_stream, apitools_request,
      apitools_download, generation=None, start_byte=0, end_byte=None,
      serialization_data=None):
    if not serialization_data:
      try:
        self.api_client.objects.Get(apitools_request,
                                    download=apitools_download)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                         object_name=object_name,
                                         generation=generation)

    # Disable apitools' default print callbacks.
    def _NoOpCallback(unused_response, unused_download_object):
      pass

    # TODO: If we have a resumable download with accept-encoding:gzip
    # on a object that is compressible but not in gzip form in the cloud,
    # on-the-fly compression will gzip the object.  In this case if our
    # download breaks, future requests will ignore the range header and just
    # return the object (gzipped) in its entirety.  Ideally, we would unzip
    # the bytes that we have locally and send a range request without
    # accept-encoding:gzip so that we can download only the (uncompressed) bytes
    # that we don't yet have.

    # Since bytes_http is created in this function, we don't get the
    # user-agent header from api_client's http automatically.
    additional_headers = {
        'accept-encoding': 'gzip',
        'user-agent': self.api_client.user_agent
    }
    if start_byte or end_byte is not None:
      apitools_download.GetRange(additional_headers=additional_headers,
                                 start=start_byte, end=end_byte,
                                 use_chunks=False)
    else:
      apitools_download.StreamMedia(
          callback=_NoOpCallback, finish_callback=_NoOpCallback,
          additional_headers=additional_headers, use_chunks=False)
    return apitools_download.encoding

  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."""
    projection = (apitools_messages.StorageObjectsPatchRequest
                  .ProjectionValueValuesEnum.full)

    if not preconditions:
      preconditions = Preconditions()

    if generation:
      generation = long(generation)

    predefined_acl = None
    apitools_include_fields = []
    if canned_acl:
      # Must null out existing ACLs to apply a canned ACL.
      apitools_include_fields.append('acl')
      predefined_acl = (
          apitools_messages.StorageObjectsPatchRequest.
          PredefinedAclValueValuesEnum(
              self._ObjectCannedAclToPredefinedAcl(canned_acl)))

    apitools_request = apitools_messages.StorageObjectsPatchRequest(
        bucket=bucket_name, object=object_name, objectResource=metadata,
        generation=generation, projection=projection,
        ifGenerationMatch=preconditions.gen_match,
        ifMetagenerationMatch=preconditions.meta_gen_match,
        predefinedAcl=predefined_acl)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    try:
      with self.api_client.IncludeFields(apitools_include_fields):
        return self.api_client.objects.Patch(apitools_request,
                                             global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _UploadObject(self, upload_stream, object_metadata, canned_acl=None,
                    size=None, preconditions=None, provider=None, fields=None,
                    serialization_data=None, tracker_callback=None,
                    progress_callback=None,
                    apitools_strategy=apitools_transfer.SIMPLE_UPLOAD,
                    total_size=0):
    # pylint: disable=g-doc-args
    """Upload implementation. Cloud API arguments, plus two more.

    Additional args:
      apitools_strategy: SIMPLE_UPLOAD or RESUMABLE_UPLOAD.
      total_size: Total size of the upload; None if it is unknown (streaming).

    Returns:
      Uploaded object metadata.
    """
    # pylint: enable=g-doc-args
    ValidateDstObjectMetadata(object_metadata)
    predefined_acl = None
    if canned_acl:
      predefined_acl = (
          apitools_messages.StorageObjectsInsertRequest.
          PredefinedAclValueValuesEnum(
              self._ObjectCannedAclToPredefinedAcl(canned_acl)))

    bytes_uploaded_container = BytesTransferredContainer()

    if progress_callback and size:
      total_size = size
      progress_callback(0, size)

    callback_class_factory = UploadCallbackConnectionClassFactory(
        bytes_uploaded_container, total_size=total_size,
        progress_callback=progress_callback)

    upload_http_class = callback_class_factory.GetConnectionClass()
    self.upload_http.connections = {'http': upload_http_class,
                                    'https': upload_http_class}

    # Since bytes_http is created in this function, we don't get the
    # user-agent header from api_client's http automatically.
    additional_headers = {
        'user-agent': self.api_client.user_agent
    }

    try:
      content_type = None
      apitools_request = None
      global_params = None
      if not serialization_data:
        # This is a new upload, set up initial upload state.
        content_type = object_metadata.contentType
        if not content_type:
          content_type = DEFAULT_CONTENT_TYPE

        if not preconditions:
          preconditions = Preconditions()

        apitools_request = apitools_messages.StorageObjectsInsertRequest(
            bucket=object_metadata.bucket, object=object_metadata,
            ifGenerationMatch=preconditions.gen_match,
            ifMetagenerationMatch=preconditions.meta_gen_match,
            predefinedAcl=predefined_acl)
        global_params = apitools_messages.StandardQueryParameters()
        if fields:
          global_params.fields = ','.join(set(fields))

      if apitools_strategy == apitools_transfer.SIMPLE_UPLOAD:
        # One-shot upload.
        apitools_upload = apitools_transfer.Upload(
            upload_stream, content_type, total_size=size, auto_transfer=True,
            num_retries=self.num_retries)
        apitools_upload.strategy = apitools_strategy
        apitools_upload.bytes_http = self.authorized_upload_http

        return self.api_client.objects.Insert(
            apitools_request,
            upload=apitools_upload,
            global_params=global_params)
      else:  # Resumable upload.
        return self._PerformResumableUpload(
            upload_stream, self.authorized_upload_http, content_type, size,
            serialization_data, apitools_strategy, apitools_request,
            global_params, bytes_uploaded_container, tracker_callback,
            additional_headers, progress_callback)
    except TRANSLATABLE_APITOOLS_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 _PerformResumableUpload(
      self, upload_stream, authorized_upload_http, content_type, size,
      serialization_data, apitools_strategy, apitools_request, global_params,
      bytes_uploaded_container, tracker_callback, addl_headers,
      progress_callback):
    try:
      if serialization_data:
        # Resuming an existing upload.
        apitools_upload = apitools_transfer.Upload.FromData(
            upload_stream, serialization_data, self.api_client.http,
            num_retries=self.num_retries)
        apitools_upload.chunksize = GetJsonResumableChunkSize()
        apitools_upload.bytes_http = authorized_upload_http
      else:
        # New resumable upload.
        apitools_upload = apitools_transfer.Upload(
            upload_stream, content_type, total_size=size,
            chunksize=GetJsonResumableChunkSize(), auto_transfer=False,
            num_retries=self.num_retries)
        apitools_upload.strategy = apitools_strategy
        apitools_upload.bytes_http = authorized_upload_http
        self.api_client.objects.Insert(
            apitools_request,
            upload=apitools_upload,
            global_params=global_params)
      # Disable retries in apitools. We will handle them explicitly here.
      apitools_upload.retry_func = (
          apitools_http_wrapper.RethrowExceptionHandler)

      # Disable apitools' default print callbacks.
      def _NoOpCallback(unused_response, unused_upload_object):
        pass

      # If we're resuming an upload, apitools has at this point received
      # from the server how many bytes it already has. Update our
      # callback class with this information.
      bytes_uploaded_container.bytes_transferred = apitools_upload.progress
      if tracker_callback:
        tracker_callback(json.dumps(apitools_upload.serialization_data))

      retries = 0
      last_progress_byte = apitools_upload.progress
      while retries <= self.num_retries:
        try:
          # TODO: On retry, this will seek to the bytes that the server has,
          # causing the hash to be recalculated. Make HashingFileUploadWrapper
          # save a digest according to json_resumable_chunk_size.
          if size:
            # If size is known, we can send it all in one request and avoid
            # making a round-trip per chunk.
            http_response = apitools_upload.StreamMedia(
                callback=_NoOpCallback, finish_callback=_NoOpCallback,
                additional_headers=addl_headers)
          else:
            # Otherwise it's a streaming request and we need to ensure that we
            # send the bytes in chunks so that we can guarantee that we never
            # need to seek backwards more than our buffer (and also that the
            # chunks are aligned to 256KB).
            http_response = apitools_upload.StreamInChunks(
                callback=_NoOpCallback, finish_callback=_NoOpCallback,
                additional_headers=addl_headers)
          processed_response = self.api_client.objects.ProcessHttpResponse(
              self.api_client.objects.GetMethodConfig('Insert'), http_response)
          if size is None and progress_callback:
            # Make final progress callback; total size should now be known.
            # This works around the fact the send function counts header bytes.
            # However, this will make the progress appear to go slightly
            # backwards at the end.
            progress_callback(apitools_upload.total_size,
                              apitools_upload.total_size)
          return processed_response
        except HTTP_TRANSFER_EXCEPTIONS, e:
          apitools_http_wrapper.RebuildHttpConnections(
              apitools_upload.bytes_http)
          while retries <= self.num_retries:
            try:
              # TODO: Simulate the refresh case in tests. Right now, our
              # mocks are not complex enough to simulate a failure.
              apitools_upload.RefreshResumableUploadState()
              start_byte = apitools_upload.progress
              bytes_uploaded_container.bytes_transferred = start_byte
              break
            except HTTP_TRANSFER_EXCEPTIONS, e2:
              apitools_http_wrapper.RebuildHttpConnections(
                  apitools_upload.bytes_http)
              retries += 1
              if retries > self.num_retries:
                raise ResumableUploadException(
                    'Transfer failed after %d retries. Final exception: %s' %
                    (self.num_retries, e2))
              time.sleep(
                  CalculateWaitForRetry(retries, max_wait=self.max_retry_wait))
          if start_byte > last_progress_byte:
            # We've made progress, so allow a fresh set of retries.
            last_progress_byte = start_byte
            retries = 0
          else:
            retries += 1
            if retries > self.num_retries:
              raise ResumableUploadException(
                  'Transfer failed after %d retries. Final exception: %s' %
                  (self.num_retries, unicode(e).encode(UTF8)))
            time.sleep(
                CalculateWaitForRetry(retries, max_wait=self.max_retry_wait))
          if self.logger.isEnabledFor(logging.DEBUG):
            self.logger.debug(
                'Retrying upload from byte %s after exception: %s. Trace: %s',
                start_byte, unicode(e).encode(UTF8), traceback.format_exc())
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      resumable_ex = self._TranslateApitoolsResumableUploadException(e)
      if resumable_ex:
        raise resumable_ex
      else:
        raise

  def UploadObject(self, upload_stream, object_metadata, canned_acl=None,
                   size=None, preconditions=None, progress_callback=None,
                   provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    return self._UploadObject(
        upload_stream, object_metadata, canned_acl=canned_acl,
        size=size, preconditions=preconditions,
        progress_callback=progress_callback, fields=fields,
        apitools_strategy=apitools_transfer.SIMPLE_UPLOAD)

  def UploadObjectStreaming(self, upload_stream, object_metadata,
                            canned_acl=None, preconditions=None,
                            progress_callback=None, provider=None,
                            fields=None):
    """See CloudApi class for function doc strings."""
    # Streaming indicated by not passing a size.
    # Resumable capabilities are present up to the resumable chunk size using
    # a buffered stream.
    return self._UploadObject(
        upload_stream, object_metadata, canned_acl=canned_acl,
        preconditions=preconditions, progress_callback=progress_callback,
        fields=fields, apitools_strategy=apitools_transfer.RESUMABLE_UPLOAD,
        total_size=None)

  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."""
    return self._UploadObject(
        upload_stream, object_metadata, canned_acl=canned_acl,
        preconditions=preconditions, fields=fields, size=size,
        serialization_data=serialization_data,
        tracker_callback=tracker_callback, progress_callback=progress_callback,
        apitools_strategy=apitools_transfer.RESUMABLE_UPLOAD)

  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."""
    ValidateDstObjectMetadata(dst_obj_metadata)
    predefined_acl = None
    if canned_acl:
      predefined_acl = (
          apitools_messages.StorageObjectsRewriteRequest.
          DestinationPredefinedAclValueValuesEnum(
              self._ObjectCannedAclToPredefinedAcl(canned_acl)))

    if src_generation:
      src_generation = long(src_generation)

    if not preconditions:
      preconditions = Preconditions()

    projection = (apitools_messages.StorageObjectsRewriteRequest.
                  ProjectionValueValuesEnum.full)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      # Rewrite returns the resultant object under the 'resource' field.
      new_fields = set(['done', 'objectSize', 'rewriteToken',
                        'totalBytesRewritten'])
      for field in fields:
        new_fields.add('resource/' + field)
      global_params.fields = ','.join(set(new_fields))

    # Check to see if we are resuming a rewrite.
    tracker_file_name = GetRewriteTrackerFilePath(
        src_obj_metadata.bucket, src_obj_metadata.name, dst_obj_metadata.bucket,
        dst_obj_metadata.name, 'JSON')
    rewrite_params_hash = HashRewriteParameters(
        src_obj_metadata, dst_obj_metadata, projection,
        src_generation=src_generation, gen_match=preconditions.gen_match,
        meta_gen_match=preconditions.meta_gen_match,
        canned_acl=predefined_acl, fields=global_params.fields,
        max_bytes_per_call=max_bytes_per_call)
    resume_rewrite_token = ReadRewriteTrackerFile(tracker_file_name,
                                                  rewrite_params_hash)

    progress_cb_with_backoff = None
    try:
      last_bytes_written = 0L
      while True:
        apitools_request = apitools_messages.StorageObjectsRewriteRequest(
            sourceBucket=src_obj_metadata.bucket,
            sourceObject=src_obj_metadata.name,
            destinationBucket=dst_obj_metadata.bucket,
            destinationObject=dst_obj_metadata.name,
            projection=projection, object=dst_obj_metadata,
            sourceGeneration=src_generation,
            ifGenerationMatch=preconditions.gen_match,
            ifMetagenerationMatch=preconditions.meta_gen_match,
            destinationPredefinedAcl=predefined_acl,
            rewriteToken=resume_rewrite_token,
            maxBytesRewrittenPerCall=max_bytes_per_call)
        rewrite_response = self.api_client.objects.Rewrite(
            apitools_request, global_params=global_params)
        bytes_written = long(rewrite_response.totalBytesRewritten)
        if progress_callback and not progress_cb_with_backoff:
          progress_cb_with_backoff = ProgressCallbackWithBackoff(
              long(rewrite_response.objectSize), progress_callback)
        if progress_cb_with_backoff:
          progress_cb_with_backoff.Progress(
              bytes_written - last_bytes_written)

        if rewrite_response.done:
          break
        elif not resume_rewrite_token:
          # Save the token and make a tracker file if they don't already exist.
          resume_rewrite_token = rewrite_response.rewriteToken
          WriteRewriteTrackerFile(tracker_file_name, rewrite_params_hash,
                                  rewrite_response.rewriteToken)
        last_bytes_written = bytes_written

      DeleteTrackerFile(tracker_file_name)
      return rewrite_response.resource
    except TRANSLATABLE_APITOOLS_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 DeleteObject(self, bucket_name, object_name, preconditions=None,
                   generation=None, provider=None):
    """See CloudApi class for function doc strings."""
    if not preconditions:
      preconditions = Preconditions()

    if generation:
      generation = long(generation)

    apitools_request = apitools_messages.StorageObjectsDeleteRequest(
        bucket=bucket_name, object=object_name, generation=generation,
        ifGenerationMatch=preconditions.gen_match,
        ifMetagenerationMatch=preconditions.meta_gen_match)
    try:
      return self.api_client.objects.Delete(apitools_request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def ComposeObject(self, src_objs_metadata, dst_obj_metadata,
                    preconditions=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    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
    if not dst_obj_metadata.contentType:
      dst_obj_metadata.contentType = DEFAULT_CONTENT_TYPE

    if not preconditions:
      preconditions = Preconditions()

    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    src_objs_compose_request = apitools_messages.ComposeRequest(
        sourceObjects=src_objs_metadata, destination=dst_obj_metadata)

    apitools_request = apitools_messages.StorageObjectsComposeRequest(
        composeRequest=src_objs_compose_request,
        destinationBucket=dst_bucket_name,
        destinationObject=dst_obj_name,
        ifGenerationMatch=preconditions.gen_match,
        ifMetagenerationMatch=preconditions.meta_gen_match)
    try:
      return self.api_client.objects.Compose(apitools_request,
                                             global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      # We can't be sure which object was missing in the 404 case.
      if isinstance(e, apitools_exceptions.HttpError) and e.status_code == 404:
        raise NotFoundException('One of the source objects does not exist.')
      else:
        self._TranslateExceptionAndRaise(e)

  def WatchBucket(self, bucket_name, address, channel_id, token=None,
                  provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageObjectsWatchAllRequest
                  .ProjectionValueValuesEnum.full)

    channel = apitools_messages.Channel(address=address, id=channel_id,
                                        token=token, type='WEB_HOOK')

    apitools_request = apitools_messages.StorageObjectsWatchAllRequest(
        bucket=bucket_name, channel=channel, projection=projection)

    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    try:
      return self.api_client.objects.WatchAll(apitools_request,
                                              global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def StopChannel(self, channel_id, resource_id, provider=None):
    """See CloudApi class for function doc strings."""
    channel = apitools_messages.Channel(id=channel_id, resourceId=resource_id)
    try:
      self.api_client.channels.Stop(channel)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS, e:
      self._TranslateExceptionAndRaise(e)

  def _BucketCannedAclToPredefinedAcl(self, canned_acl_string):
    """Translates the input string to a bucket PredefinedAcl string.

    Args:
      canned_acl_string: Canned ACL string.

    Returns:
      String that can be used as a query parameter with the JSON API. This
      corresponds to a flavor of *PredefinedAclValueValuesEnum and can be
      used as input to apitools requests that affect bucket access controls.
    """
    # XML : JSON
    translation_dict = {
        None: None,
        'authenticated-read': 'authenticatedRead',
        'private': 'private',
        'project-private': 'projectPrivate',
        'public-read': 'publicRead',
        'public-read-write': 'publicReadWrite'
    }
    if canned_acl_string in translation_dict:
      return translation_dict[canned_acl_string]
    raise ArgumentException('Invalid canned ACL %s' % canned_acl_string)

  def _ObjectCannedAclToPredefinedAcl(self, canned_acl_string):
    """Translates the input string to an object PredefinedAcl string.

    Args:
      canned_acl_string: Canned ACL string.

    Returns:
      String that can be used as a query parameter with the JSON API. This
      corresponds to a flavor of *PredefinedAclValueValuesEnum and can be
      used as input to apitools requests that affect object access controls.
    """
    # XML : JSON
    translation_dict = {
        None: None,
        'authenticated-read': 'authenticatedRead',
        'bucket-owner-read': 'bucketOwnerRead',
        'bucket-owner-full-control': 'bucketOwnerFullControl',
        'private': 'private',
        'project-private': 'projectPrivate',
        'public-read': 'publicRead'
    }
    if canned_acl_string in translation_dict:
      return translation_dict[canned_acl_string]
    raise ArgumentException('Invalid canned ACL %s' % canned_acl_string)

  def _TranslateExceptionAndRaise(self, e, bucket_name=None, object_name=None,
                                  generation=None, not_found_exception=None):
    """Translates an HTTP 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._TranslateApitoolsException(
        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 _GetMessageFromHttpError(self, http_error):
    if isinstance(http_error, apitools_exceptions.HttpError):
      if getattr(http_error, 'content', None):
        try:
          json_obj = json.loads(http_error.content)
          if 'error' in json_obj and 'message' in json_obj['error']:
            return json_obj['error']['message']
        except Exception:  # pylint: disable=broad-except
          # If we couldn't decode anything, just leave the message as None.
          pass

  def _TranslateApitoolsResumableUploadException(self, e):
    if isinstance(e, apitools_exceptions.HttpError):
      message = self._GetMessageFromHttpError(e)
      if (e.status_code == 503 and
          self.http.disable_ssl_certificate_validation):
        return ServiceException(_VALIDATE_CERTIFICATES_503_MESSAGE,
                                status=e.status_code)
      elif e.status_code >= 500:
        return ResumableUploadException(
            message or 'Server Error', status=e.status_code)
      elif e.status_code == 429:
        return ResumableUploadException(
            message or 'Too Many Requests', status=e.status_code)
      elif e.status_code == 410:
        return ResumableUploadStartOverException(
            message or 'Bad Request', status=e.status_code)
      elif e.status_code == 404:
        return ResumableUploadStartOverException(
            message or 'Bad Request', status=e.status_code)
      elif e.status_code >= 400:
        return ResumableUploadAbortException(
            message or 'Bad Request', status=e.status_code)
    if isinstance(e, apitools_exceptions.StreamExhausted):
      return ResumableUploadAbortException(e.message)
    if (isinstance(e, apitools_exceptions.TransferError) and
        ('Aborting transfer' in e.message or
         'Not enough bytes in stream' in e.message or
         'additional bytes left in stream' in e.message)):
      return ResumableUploadAbortException(e.message)

  def _TranslateApitoolsException(self, e, bucket_name=None, object_name=None,
                                  generation=None, not_found_exception=None):
    """Translates apitools exceptions into their gsutil Cloud Api equivalents.

    Args:
      e: Any exception in TRANSLATABLE_APITOOLS_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.
    """
    if isinstance(e, apitools_exceptions.HttpError):
      message = self._GetMessageFromHttpError(e)
      if e.status_code == 400:
        # It is possible that the Project ID is incorrect.  Unfortunately the
        # JSON API does not give us much information about what part of the
        # request was bad.
        return BadRequestException(message or 'Bad Request',
                                   status=e.status_code)
      elif e.status_code == 401:
        if 'Login Required' in str(e):
          return AccessDeniedException(
              message or 'Access denied: login required.',
              status=e.status_code)
      elif e.status_code == 403:
        if 'The account for the specified project has been disabled' in str(e):
          return AccessDeniedException(message or 'Account disabled.',
                                       status=e.status_code)
        elif 'Daily Limit for Unauthenticated Use Exceeded' in str(e):
          return AccessDeniedException(
              message or 'Access denied: quota exceeded. '
              'Is your project ID valid?',
              status=e.status_code)
        elif 'The bucket you tried to delete was not empty.' in str(e):
          return NotEmptyException('BucketNotEmpty (%s)' % bucket_name,
                                   status=e.status_code)
        elif ('The bucket you tried to create requires domain ownership '
              'verification.' in str(e)):
          return AccessDeniedException(
              'The bucket you tried to create requires domain ownership '
              'verification. Please see '
              'https://developers.google.com/storage/docs/bucketnaming'
              '?hl=en#verification for more details.', status=e.status_code)
        elif 'User Rate Limit Exceeded' in str(e):
          return AccessDeniedException('Rate limit exceeded. Please retry this '
                                       'request later.', status=e.status_code)
        elif 'Access Not Configured' in str(e):
          return AccessDeniedException(
              'Access Not Configured. Please go to the Google Developers '
              'Console (https://cloud.google.com/console#/project) for your '
              'project, select APIs and Auth and enable the '
              'Google Cloud Storage JSON API.',
              status=e.status_code)
        else:
          return AccessDeniedException(message or e.message,
                                       status=e.status_code)
      elif e.status_code == 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_code)
          return not_found_exception
        elif bucket_name:
          if object_name:
            return CreateObjectNotFoundException(e.status_code, self.provider,
                                                 bucket_name, object_name,
                                                 generation=generation)
          return CreateBucketNotFoundException(e.status_code, self.provider,
                                               bucket_name)
        return NotFoundException(e.message, status=e.status_code)

      elif e.status_code == 409 and bucket_name:
        if 'The bucket you tried to delete was not empty.' in str(e):
          return NotEmptyException('BucketNotEmpty (%s)' % bucket_name,
                                   status=e.status_code)
        return ServiceException(
            'Bucket %s already exists.' % bucket_name, status=e.status_code)
      elif e.status_code == 412:
        return PreconditionException(message, status=e.status_code)
      elif (e.status_code == 503 and
            not self.http.disable_ssl_certificate_validation):
        return ServiceException(_VALIDATE_CERTIFICATES_503_MESSAGE,
                                status=e.status_code)
      return ServiceException(message, status=e.status_code)
    elif isinstance(e, apitools_exceptions.TransferInvalidError):
      return ServiceException('Transfer invalid (possible encoding error: %s)'
                              % str(e))