Pass 'user_project' if set for Blob API requests (#3495)

Author: tseaver

storage/google/cloud/storage/blob.py

@@ -340,10 +340,14 @@ def exists(self, client=None):
         :returns: True if the blob exists in Cloud Storage.
         """
         client = self._require_client(client)
+        # We only need the status code (200 or not) so we seek to
+        # minimize the returned payload.
+        query_params = {'fields': 'name'}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         try:
-            # We only need the status code (200 or not) so we seek to
-            # minimize the returned payload.
-            query_params = {'fields': 'name'}
             # We intentionally pass `_target_object=None` since fields=name
             # would limit the local properties.
             client._connection.api_request(
@@ -403,6 +407,8 @@ def _get_download_url(self):
             download_url = _DOWNLOAD_URL_TEMPLATE.format(path=self.path)
             if self.generation is not None:
                 download_url += u'&generation={:d}'.format(self.generation)
+            if self.user_project is not None:
+                download_url += u'&userProject={}'.format(self.user_project)
             return download_url
         else:
             return self.media_link
@@ -654,6 +660,10 @@ def _do_multipart_upload(self, client, stream, content_type,
 
         upload_url = _MULTIPART_URL_TEMPLATE.format(
             bucket_path=self.bucket.path)
+
+        if self.user_project is not None:
+            upload_url += '&userProject={}'.format(self.user_project)
+
         upload = MultipartUpload(upload_url, headers=headers)
 
         if num_retries is not None:
@@ -726,6 +736,10 @@ def _initiate_resumable_upload(self, client, stream, content_type,
 
         upload_url = _RESUMABLE_URL_TEMPLATE.format(
             bucket_path=self.bucket.path)
+
+        if self.user_project is not None:
+            upload_url += '&userProject={}'.format(self.user_project)
+
         upload = ResumableUpload(upload_url, chunk_size, headers=headers)
 
         if num_retries is not None:
@@ -1079,9 +1093,16 @@ def get_iam_policy(self, client=None):
                   the ``getIamPolicy`` API request.
         """
         client = self._require_client(client)
+
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         info = client._connection.api_request(
             method='GET',
             path='%s/iam' % (self.path,),
+            query_params=query_params,
             _target_object=None)
         return Policy.from_api_repr(info)
 
@@ -1104,11 +1125,18 @@ def set_iam_policy(self, policy, client=None):
                   the ``setIamPolicy`` API request.
         """
         client = self._require_client(client)
+
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         resource = policy.to_api_repr()
         resource['resourceId'] = self.path
         info = client._connection.api_request(
             method='PUT',
             path='%s/iam' % (self.path,),
+            query_params=query_params,
             data=resource,
             _target_object=None)
         return Policy.from_api_repr(info)
@@ -1132,12 +1160,17 @@ def test_iam_permissions(self, permissions, client=None):
                   request.
         """
         client = self._require_client(client)
-        query = {'permissions': permissions}
+        query_params = {'permissions': permissions}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         path = '%s/iam/testPermissions' % (self.path,)
         resp = client._connection.api_request(
             method='GET',
             path=path,
-            query_params=query)
+            query_params=query_params)
+
         return resp.get('permissions', [])
 
     def make_public(self, client=None):
@@ -1167,13 +1200,22 @@ def compose(self, sources, client=None):
         """
         if self.content_type is None:
             raise ValueError("Destination 'content_type' not set.")
+
         client = self._require_client(client)
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         request = {
             'sourceObjects': [{'name': source.name} for source in sources],
             'destination': self._properties.copy(),
         }
         api_response = client._connection.api_request(
-            method='POST', path=self.path + '/compose', data=request,
+            method='POST',
+            path=self.path + '/compose',
+            query_params=query_params,
+            data=request,
             _target_object=self)
         self._set_properties(api_response)
 
@@ -1205,14 +1247,20 @@ def rewrite(self, source, token=None, client=None):
         headers.update(_get_encryption_headers(
             source._encryption_key, source=True))
 
+        query_params = {}
+
         if token:
-            query_params = {'rewriteToken': token}
-        else:
-            query_params = {}
+            query_params['rewriteToken'] = token
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
 
         api_response = client._connection.api_request(
-            method='POST', path=source.path + '/rewriteTo' + self.path,
-            query_params=query_params, data=self._properties, headers=headers,
+            method='POST',
+            path=source.path + '/rewriteTo' + self.path,
+            query_params=query_params,
+            data=self._properties,
+            headers=headers,
             _target_object=self)
         rewritten = int(api_response['totalBytesRewritten'])
         size = int(api_response['objectSize'])
@@ -1243,13 +1291,22 @@ def update_storage_class(self, new_class, client=None):
             raise ValueError("Invalid storage class: %s" % (new_class,))
 
         client = self._require_client(client)
+
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         headers = _get_encryption_headers(self._encryption_key)
         headers.update(_get_encryption_headers(
             self._encryption_key, source=True))
 
         api_response = client._connection.api_request(
-            method='POST', path=self.path + '/rewriteTo' + self.path,
-            data={'storageClass': new_class}, headers=headers,
+            method='POST',
+            path=self.path + '/rewriteTo' + self.path,
+            query_params=query_params,
+            data={'storageClass': new_class},
+            headers=headers,
             _target_object=self)
         self._set_properties(api_response['resource'])