Pass 'user_project' if set for Bucket API requests (#3492)

* Block 'Bucket.create' if 'user_project' set:  the API does not accept that parameter.

storage/google/cloud/storage/bucket.py

@@ -175,10 +175,14 @@ def exists(self, client=None):
         :returns: True if the bucket 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(
@@ -204,6 +208,9 @@ def create(self, client=None):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
         """
+        if self.user_project is not None:
+            raise ValueError("Cannot create bucket with 'user_project' set.")
+
         client = self._require_client(client)
         query_params = {'project': client.project}
         properties = {key: self._properties[key] for key in self._changes}
@@ -264,10 +271,18 @@ def get_blob(self, blob_name, client=None):
         :returns: The blob object if it exists, otherwise None.
         """
         client = self._require_client(client)
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         blob = Blob(bucket=self, name=blob_name)
         try:
             response = client._connection.api_request(
-                method='GET', path=blob.path, _target_object=blob)
+                method='GET',
+                path=blob.path,
+                query_params=query_params,
+                _target_object=blob)
             # NOTE: We assume response.get('name') matches `blob_name`.
             blob._set_properties(response)
             # NOTE: This will not fail immediately in a batch. However, when
@@ -321,7 +336,7 @@ def list_blobs(self, max_results=None, page_token=None, prefix=None,
         :returns: Iterator of all :class:`~google.cloud.storage.blob.Blob`
                   in this bucket matching the arguments.
         """
-        extra_params = {}
+        extra_params = {'projection': projection}
 
         if prefix is not None:
             extra_params['prefix'] = prefix
@@ -332,11 +347,12 @@ def list_blobs(self, max_results=None, page_token=None, prefix=None,
         if versions is not None:
             extra_params['versions'] = versions
 
-        extra_params['projection'] = projection
-
         if fields is not None:
             extra_params['fields'] = fields
 
+        if self.user_project is not None:
+            extra_params['userProject'] = self.user_project
+
         client = self._require_client(client)
         path = self.path + '/o'
         iterator = HTTPIterator(
@@ -376,6 +392,11 @@ def delete(self, force=False, client=None):
                  contains more than 256 objects / blobs.
         """
         client = self._require_client(client)
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         if force:
             blobs = list(self.list_blobs(
                 max_results=self._MAX_OBJECTS_FOR_ITERATION + 1,
@@ -397,7 +418,10 @@ def delete(self, force=False, client=None):
         # request has no response value (whether in a standard request or
         # in a batch request).
         client._connection.api_request(
-            method='DELETE', path=self.path, _target_object=None)
+            method='DELETE',
+            path=self.path,
+            query_params=query_params,
+            _target_object=None)
 
     def delete_blob(self, blob_name, client=None):
         """Deletes a blob from the current bucket.
@@ -429,12 +453,20 @@ def delete_blob(self, blob_name, client=None):
 
         """
         client = self._require_client(client)
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         blob_path = Blob.path_helper(self.path, blob_name)
         # We intentionally pass `_target_object=None` since a DELETE
         # request has no response value (whether in a standard request or
         # in a batch request).
         client._connection.api_request(
-            method='DELETE', path=blob_path, _target_object=None)
+            method='DELETE',
+            path=blob_path,
+            query_params=query_params,
+            _target_object=None)
 
     def delete_blobs(self, blobs, on_error=None, client=None):
         """Deletes a list of blobs from the current bucket.
@@ -497,14 +529,26 @@ def copy_blob(self, blob, destination_bucket, new_name=None,
         :returns: The new Blob.
         """
         client = self._require_client(client)
+        query_params = {}
+
+        if self.user_project is not None:
+            query_params['userProject'] = self.user_project
+
         if new_name is None:
             new_name = blob.name
+
         new_blob = Blob(bucket=destination_bucket, name=new_name)
         api_path = blob.path + '/copyTo' + new_blob.path
         copy_result = client._connection.api_request(
-            method='POST', path=api_path, _target_object=new_blob)
+            method='POST',
+            path=api_path,
+            query_params=query_params,
+            _target_object=new_blob,
+            )
+
         if not preserve_acl:
             new_blob.acl.save(acl={}, client=client)
+
         new_blob._set_properties(copy_result)
         return new_blob
 
@@ -912,9 +956,15 @@ 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)
 
@@ -937,11 +987,17 @@ 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)
@@ -965,12 +1021,16 @@ 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, recursive=False, future=False, client=None):