feat: add support for partial list buckets

google/cloud/storage/client.py

@@ -64,6 +64,12 @@
 _marker = object()
 
 
+def _buckets_page_start(iterator, page, response):
+    """Grab unreachable buckets after a :class:`~google.cloud.iterator.Page` started."""
+    unreachable = response.get("unreachable", None)
+    if unreachable:
+        iterator.unreachable.extend(unreachable)
+
 class Client(ClientWithProject):
     """Client to bundle configuration needed for API requests.
 
@@ -1458,6 +1464,7 @@ def list_buckets(
         retry=DEFAULT_RETRY,
         *,
         soft_deleted=None,
+        return_partial_success=None,
     ):
         """Get all buckets in the project associated to the client.
 
@@ -1516,6 +1523,13 @@ def list_buckets(
             generation number. This parameter can only be used successfully if the bucket has a soft delete policy.
             See: https://cloud.google.com/storage/docs/soft-delete
 
+        :type return_partial_success: bool
+        :param return_partial_success:
+            (Optional) If True, the response will also contain a list of
+            unreachable buckets if the buckets are unavailable. The
+            unreachable buckets will be available on the ``unreachable``
+            attribute of the returned iterator.
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :raises ValueError: if both ``project`` is ``None`` and the client's
                             project is also ``None``.
@@ -1551,16 +1565,34 @@ def list_buckets(
             if soft_deleted is not None:
                 extra_params["softDeleted"] = soft_deleted
 
-            return self._list_resource(
-                "/b",
-                _item_to_bucket,
-                page_token=page_token,
-                max_results=max_results,
-                extra_params=extra_params,
-                page_size=page_size,
-                timeout=timeout,
-                retry=retry,
-            )
+            if return_partial_success is not None:
+                extra_params["returnPartialSuccess"] = return_partial_success
+
+                iterator = self._list_resource(
+                    "/b",
+                    _item_to_bucket,
+                    page_token=page_token,
+                    max_results=max_results,
+                    extra_params=extra_params,
+                    page_size=page_size,
+                    timeout=timeout,
+                    retry=retry,
+                    page_start=_buckets_page_start,
+                )
+                iterator.unreachable = []
+
+            else:
+                iterator = self._list_resource(
+                    "/b",
+                    _item_to_bucket,
+                    page_token=page_token,
+                    max_results=max_results,
+                    extra_params=extra_params,
+                    page_size=page_size,
+                    timeout=timeout,
+                    retry=retry,
+                )
+            return iterator
 
     def restore_bucket(
         self,

tests/unit/test_client.py

@@ -3086,6 +3086,39 @@ def test_get_signed_policy_v4_with_access_token_sa_email(self):
         self.assertEqual(fields["x-goog-signature"], EXPECTED_SIGN)
         self.assertEqual(fields["policy"], EXPECTED_POLICY)
 
+    def test_list_buckets_w_partial_success(self):
+        from google.cloud.storage.client import _item_to_bucket
+        from google.cloud.storage.client import _buckets_page_start
+
+        PROJECT = "PROJECT"
+        client = self._make_one(project=PROJECT)
+        mock_iterator = mock.Mock()
+        client._list_resource = mock.Mock(return_value=mock_iterator)
+
+        iterator = client.list_buckets(return_partial_success=True)
+
+        self.assertIs(iterator, mock_iterator)
+        self.assertEqual(iterator.unreachable, [])
+
+        expected_path = "/b"
+        expected_item_to_value = _item_to_bucket
+        expected_extra_params = {
+            "project": PROJECT,
+            "projection": "noAcl",
+            "returnPartialSuccess": True,
+        }
+
+        client._list_resource.assert_called_once_with(
+            expected_path,
+            expected_item_to_value,
+            page_token=None,
+            max_results=None,
+            extra_params=expected_extra_params,
+            page_size=None,
+            timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
+            page_start=_buckets_page_start,
+        )
 
 class Test__item_to_bucket(unittest.TestCase):
     def _call_fut(self, iterator, item):