Roll version of gsutil to 4.15.

third_party/gsutil/gslib/gcs_json_api.py

 # -*- 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
@@ skipping 18 matching lines @@
 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
@@ skipping 17 matching lines @@
 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
@@ skipping 37 matching lines @@
     """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):
+               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 '
@@ skipping 12 matching lines @@
             % 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_string_key(
-            GetCredentialStoreFilename(), self.api_version))
+        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)
+        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 = []
@@ skipping 81 matching lines @@
         raise
 
   def _GetDevshellCreds(self):
     try:
       return devshell.DevshellCredentials()
     except devshell.NoDevshellServer:
       return None
     except:
       raise
 
-  def _GetNewDownloadHttp(self, download_stream):
-    return GetNewHttp(http_class=HttpWithDownloadStream, stream=download_stream)
+  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(
@@ skipping 29 matching lines @@
                            '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
@@ skipping 148 matching lines @@
       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)
-    if object_list.prefixes:
-      for prefix in object_list.prefixes:
-        yield CloudApi.CsObjectOrPrefix(prefix,
-                                        CloudApi.CsObjectOrPrefixType.PREFIX)
 
   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)
 
@@ skipping 17 matching lines @@
       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 serialization_data:
+    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(0, outer_total_size)
+      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()
 
-    download_http = self._GetNewDownloadHttp(download_stream)
-    download_http.connections = {'https': download_http_class}
-    authorized_download_http = self.credentials.authorize(download_http)
-    WrapDownloadHttpRequest(authorized_download_http)
+    # 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 = authorized_download_http
+    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,
@@ skipping 27 matching lines @@
         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=GetMaxRetryDelay()))
+        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,
@@ skipping 19 matching lines @@
     # 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:
+    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,
@@ skipping 66 matching lines @@
     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 = self._GetNewUploadHttp()
     upload_http_class = callback_class_factory.GetConnectionClass()
-    upload_http.connections = {'http': upload_http_class,
-                               'https': upload_http_class}
+    self.upload_http.connections = {'http': upload_http_class,
+                                    'https': upload_http_class}
 
-    authorized_upload_http = self.credentials.authorize(upload_http)
-    WrapUploadHttpRequest(authorized_upload_http)
     # 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
@@ skipping 14 matching lines @@
         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 = authorized_upload_http
+        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, authorized_upload_http, content_type, size,
+            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)
+                                       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(
@@ skipping 72 matching lines @@
               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=GetMaxRetryDelay()))
+                  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=GetMaxRetryDelay()))
+                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
@@ skipping 107 matching lines @@
         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)
+                                       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)
 
@@ skipping 119 matching lines @@
         '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):
+                                  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)
+        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, bucket_name=None, object_name=None, generation=None):
+  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:
@@ skipping 10 matching lines @@
             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):
+                                  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
@@ skipping 31 matching lines @@
           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 bucket_name:
+        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))