Repository: googleapis/python-storage

Issue:
Add a default retry parameter to all public methods
All public storage method should have a default retry parametr defined so that clients don't have to explicitly use the `@retry.Retry` decorator to handle transient issues, like 503 error code.

The solution could look similar to the implementation in the [google-cloud-bigquery](https://github.com/googleapis/python-bigquery/blob/master/google/cloud/bigquery/client.py#L830) where retry is implemented in every public method and it is set to DEFAULT_RETRY by default.

Discussion:
andrewsg: Thanks! This is a major change to the UI in line with our medium-term plan for retry support, so I (author of underlying retry logic) will review thoroughly soon.
andrewsg: @HemangChothani are we clear to merge?
HemangChothani: @andrewsg Yes.

Patch:
diff --git a/google/cloud/storage/_helpers.py b/google/cloud/storage/_helpers.py
index ba59f8f..338b798 100644
--- a/google/cloud/storage/_helpers.py
+++ b/google/cloud/storage/_helpers.py
@@ -145,6 +145,7 @@ class _PropertyMixin(object):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY,
     ):
         """Reload properties from Cloud Storage.
 
@@ -187,6 +188,20 @@ class _PropertyMixin(object):
         :type if_metageneration_not_match: long
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         client = self._require_client(client)
         query_params = self._query_params
@@ -207,7 +222,7 @@ class _PropertyMixin(object):
             headers=self._encryption_headers(),
             _target_object=self,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
         self._set_properties(api_response)
 
@@ -247,6 +262,7 @@ class _PropertyMixin(object):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """Sends all changed properties in a PATCH request.
 
@@ -286,6 +302,20 @@ class _PropertyMixin(object):
         :type if_metageneration_not_match: long
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         client = self._require_client(client)
         query_params = self._query_params
@@ -309,7 +339,7 @@ class _PropertyMixin(object):
             query_params=query_params,
             _target_object=self,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
+            retry=retry,
         )
         self._set_properties(api_response)
 
@@ -321,6 +351,7 @@ class _PropertyMixin(object):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """Sends all properties in a PUT request.
 
@@ -360,6 +391,20 @@ class _PropertyMixin(object):
         :type if_metageneration_not_match: long
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         client = self._require_client(client)
 
@@ -380,7 +425,7 @@ class _PropertyMixin(object):
             query_params=query_params,
             _target_object=self,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
+            retry=retry,
         )
         self._set_properties(api_response)
 
diff --git a/google/cloud/storage/acl.py b/google/cloud/storage/acl.py
index 765590f..55c12c9 100644
--- a/google/cloud/storage/acl.py
+++ b/google/cloud/storage/acl.py
@@ -456,7 +456,7 @@ class ACL(object):
         self.entities.clear()
 
         found = client._connection.api_request(
-            method="GET", path=path, query_params=query_params, timeout=timeout
+            method="GET", path=path, query_params=query_params, timeout=timeout,
         )
         self.loaded = True
         for entry in found.get("items", ()):
diff --git a/google/cloud/storage/blob.py b/google/cloud/storage/blob.py
index ef7b88b..044ca49 100644
--- a/google/cloud/storage/blob.py
+++ b/google/cloud/storage/blob.py
@@ -75,6 +75,8 @@ from google.cloud.storage.constants import MULTI_REGIONAL_LEGACY_STORAGE_CLASS
 from google.cloud.storage.constants import NEARLINE_STORAGE_CLASS
 from google.cloud.storage.constants import REGIONAL_LEGACY_STORAGE_CLASS
 from google.cloud.storage.constants import STANDARD_STORAGE_CLASS
+from google.cloud.storage.retry import DEFAULT_RETRY
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
 
 
@@ -595,6 +597,7 @@ class Blob(_PropertyMixin):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Determines whether or not this blob exists.
 
@@ -637,6 +640,20 @@ class Blob(_PropertyMixin):
             (Optional) Make the operation conditional on whether the blob's
             current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: bool
         :returns: True if the blob exists in Cloud Storage.
         """
@@ -662,6 +679,7 @@ class Blob(_PropertyMixin):
                 query_params=query_params,
                 _target_object=None,
                 timeout=timeout,
+                retry=retry,
             )
             # NOTE: This will not fail immediately in a batch. However, when
             #       Batch.finish() is called, the resulting `NotFound` will be
@@ -678,6 +696,7 @@ class Blob(_PropertyMixin):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Deletes a blob from Cloud Storage.
 
@@ -720,6 +739,20 @@ class Blob(_PropertyMixin):
             (Optional) Make the operation conditional on whether the blob's
             current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises: :class:`google.cloud.exceptions.NotFound`
                  (propagated from
                  :meth:`google.cloud.storage.bucket.Bucket.delete_blob`).
@@ -733,6 +766,7 @@ class Blob(_PropertyMixin):
             if_generation_not_match=if_generation_not_match,
             if_metageneration_match=if_metageneration_match,
             if_metageneration_not_match=if_metageneration_not_match,
+            retry=retry,
         )
 
     def _get_transport(self, client):
@@ -2678,7 +2712,11 @@ class Blob(_PropertyMixin):
             _raise_from_invalid_response(exc)
 
     def get_iam_policy(
-        self, client=None, requested_policy_version=None, timeout=_DEFAULT_TIMEOUT
+        self,
+        client=None,
+        requested_policy_version=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Retrieve the IAM policy for the object.
 
@@ -2717,6 +2755,20 @@ class Blob(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
                   the ``getIamPolicy`` API request.
@@ -2737,10 +2789,17 @@ class Blob(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
+            retry=retry,
         )
         return Policy.from_api_repr(info)
 
-    def set_iam_policy(self, policy, client=None, timeout=_DEFAULT_TIMEOUT):
+    def set_iam_policy(
+        self,
+        policy,
+        client=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY_IF_ETAG_IN_JSON,
+    ):
         """Update the IAM policy for the bucket.
 
         .. note:
@@ -2770,6 +2829,20 @@ class Blob(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
                   the ``setIamPolicy`` API request.
@@ -2790,10 +2863,13 @@ class Blob(_PropertyMixin):
             data=resource,
             _target_object=None,
             timeout=timeout,
+            retry=retry,
         )
         return Policy.from_api_repr(info)
 
-    def test_iam_permissions(self, permissions, client=None, timeout=_DEFAULT_TIMEOUT):
+    def test_iam_permissions(
+        self, permissions, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+    ):
         """API call:  test permissions
 
         .. note:
@@ -2823,6 +2899,20 @@ class Blob(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: list of string
         :returns: the permissions returned by the ``testIamPermissions`` API
                   request.
@@ -2835,7 +2925,11 @@ class Blob(_PropertyMixin):
 
         path = "%s/iam/testPermissions" % (self.path,)
         resp = client._connection.api_request(
-            method="GET", path=path, query_params=query_params, timeout=timeout
+            method="GET",
+            path=path,
+            query_params=query_params,
+            timeout=timeout,
+            retry=retry,
         )
 
         return resp.get("permissions", [])
@@ -2843,10 +2937,10 @@ class Blob(_PropertyMixin):
     def make_public(self, client=None):
         """Update blob's ACL, granting read access to anonymous users.
 
-        :type client: :class:`~google.cloud.storage.client.Client`
-        :param client:
-            (Optional) The client to use.  If not passed, falls back to the
-            ``client`` stored on the blob's bucket.
+        :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+        :param client: (Optional) The client to use.  If not passed, falls back
+                       to the ``client`` stored on the blob's bucket.
         """
         self.acl.all().grant_read()
         self.acl.save(client=client)
@@ -2854,10 +2948,10 @@ class Blob(_PropertyMixin):
     def make_private(self, client=None):
         """Update blob's ACL, revoking read access for anonymous users.
 
-        :type client: :class:`~google.cloud.storage.client.Client`
-        :param client:
-            (Optional) The client to use.  If not passed, falls back to the
-            ``client`` stored on the blob's bucket.
+        :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+        :param client: (Optional) The client to use.  If not passed, falls back
+                       to the ``client`` stored on the blob's bucket.
         """
         self.acl.all().revoke_read()
         self.acl.save(client=client)
@@ -2869,6 +2963,7 @@ class Blob(_PropertyMixin):
         timeout=_DEFAULT_TIMEOUT,
         if_generation_match=None,
         if_metageneration_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Concatenate source blobs into this one.
 
@@ -2903,6 +2998,20 @@ class Blob(_PropertyMixin):
             current metageneration matches the given value. The list must match
             ``sources`` item-to-item.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         Example:
             Compose blobs using generation match preconditions.
 
@@ -2970,7 +3079,7 @@ class Blob(_PropertyMixin):
             data=request,
             _target_object=self,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            retry=retry,
         )
         self._set_properties(api_response)
 
@@ -2988,6 +3097,7 @@ class Blob(_PropertyMixin):
         if_source_generation_not_match=None,
         if_source_metageneration_match=None,
         if_source_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Rewrite source blob into this one.
 
@@ -3063,6 +3173,20 @@ class Blob(_PropertyMixin):
             (Optional) Makes the operation conditional on whether the source
             object's current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: tuple
         :returns: ``(token, bytes_rewritten, total_bytes)``, where ``token``
                   is a rewrite token (``None`` if the rewrite is complete),
@@ -3107,7 +3231,7 @@ class Blob(_PropertyMixin):
             headers=headers,
             _target_object=self,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            retry=retry,
         )
         rewritten = int(api_response["totalBytesRewritten"])
         size = int(api_response["objectSize"])
@@ -3134,6 +3258,7 @@ class Blob(_PropertyMixin):
         if_source_metageneration_match=None,
         if_source_metageneration_not_match=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Update blob's storage class via a rewrite-in-place. This helper will
         wait for the rewrite to complete before returning, so it may take some
@@ -3215,6 +3340,20 @@ class Blob(_PropertyMixin):
             repeated several times using the same timeout each time.
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         if new_class not in self.STORAGE_CLASSES:
             raise ValueError("Invalid storage class: %s" % (new_class,))
@@ -3234,6 +3373,7 @@ class Blob(_PropertyMixin):
             if_source_metageneration_match=if_source_metageneration_match,
             if_source_metageneration_not_match=if_source_metageneration_not_match,
             timeout=timeout,
+            retry=retry,
         )
         while token is not None:
             token, _, _ = self.rewrite(
@@ -3248,6 +3388,7 @@ class Blob(_PropertyMixin):
                 if_source_metageneration_match=if_source_metageneration_match,
                 if_source_metageneration_not_match=if_source_metageneration_not_match,
                 timeout=timeout,
+                retry=retry,
             )
 
     cache_control = _scalar_property("cacheControl")
diff --git a/google/cloud/storage/bucket.py b/google/cloud/storage/bucket.py
index 7ab9a13..8fb2881 100644
--- a/google/cloud/storage/bucket.py
+++ b/google/cloud/storage/bucket.py
@@ -60,6 +60,8 @@ from google.cloud.storage.notification import NONE_PAYLOAD_FORMAT
 from google.cloud.storage.retry import DEFAULT_RETRY
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+
 
 _UBLA_BPO_ENABLED_MESSAGE = (
     "Pass only one of 'uniform_bucket_level_access_enabled' / "
@@ -721,6 +723,7 @@ class Bucket(_PropertyMixin):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """Determines whether or not this bucket exists.
 
@@ -746,6 +749,20 @@ class Bucket(_PropertyMixin):
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: bool
         :returns: True if the bucket exists in Cloud Storage.
         """
@@ -771,6 +788,7 @@ class Bucket(_PropertyMixin):
                 query_params=query_params,
                 _target_object=None,
                 timeout=timeout,
+                retry=retry,
             )
             # NOTE: This will not fail immediately in a batch. However, when
             #       Batch.finish() is called, the resulting `NotFound` will be
@@ -787,6 +805,7 @@ class Bucket(_PropertyMixin):
         predefined_acl=None,
         predefined_default_object_acl=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """DEPRECATED. Creates current bucket.
 
@@ -831,6 +850,20 @@ class Bucket(_PropertyMixin):
 
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         warnings.warn(
             "Bucket.create() is deprecated and will be removed in future."
@@ -849,6 +882,7 @@ class Bucket(_PropertyMixin):
             predefined_acl=predefined_acl,
             predefined_default_object_acl=predefined_default_object_acl,
             timeout=timeout,
+            retry=retry,
         )
 
     def update(
@@ -857,6 +891,7 @@ class Bucket(_PropertyMixin):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """Sends all properties in a PUT request.
 
@@ -883,12 +918,27 @@ class Bucket(_PropertyMixin):
         :type if_metageneration_not_match: long
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         super(Bucket, self).update(
             client=client,
             timeout=timeout,
             if_metageneration_match=if_metageneration_match,
             if_metageneration_not_match=if_metageneration_not_match,
+            retry=retry,
         )
 
     def reload(
@@ -898,6 +948,7 @@ class Bucket(_PropertyMixin):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY,
     ):
         """Reload properties from Cloud Storage.
 
@@ -927,6 +978,20 @@ class Bucket(_PropertyMixin):
         :type if_metageneration_not_match: long
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         super(Bucket, self).reload(
             client=client,
@@ -934,6 +999,7 @@ class Bucket(_PropertyMixin):
             timeout=timeout,
             if_metageneration_match=if_metageneration_match,
             if_metageneration_not_match=if_metageneration_not_match,
+            retry=retry,
         )
 
     def patch(
@@ -942,6 +1008,7 @@ class Bucket(_PropertyMixin):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """Sends all changed properties in a PATCH request.
 
@@ -968,6 +1035,20 @@ class Bucket(_PropertyMixin):
         :type if_metageneration_not_match: long
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
+
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
         """
         # Special case: For buckets, it is possible that labels are being
         # removed; this requires special handling.
@@ -983,6 +1064,7 @@ class Bucket(_PropertyMixin):
             timeout=timeout,
             if_metageneration_match=if_metageneration_match,
             if_metageneration_not_match=if_metageneration_not_match,
+            retry=retry,
         )
 
     @property
@@ -1026,6 +1108,7 @@ class Bucket(_PropertyMixin):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
         **kwargs
     ):
         """Get a blob object by name.
@@ -1085,6 +1168,20 @@ class Bucket(_PropertyMixin):
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :param kwargs: Keyword arguments to pass to the
                        :class:`~google.cloud.storage.blob.Blob` constructor.
 
@@ -1109,6 +1206,7 @@ class Bucket(_PropertyMixin):
                 if_generation_not_match=if_generation_not_match,
                 if_metageneration_match=if_metageneration_match,
                 if_metageneration_not_match=if_metageneration_not_match,
+                retry=retry,
             )
         except NotFound:
             return None
@@ -1129,6 +1227,7 @@ class Bucket(_PropertyMixin):
         fields=None,
         client=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -1205,6 +1304,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :returns: Iterator of all :class:`~google.cloud.storage.blob.Blob`
                   in this bucket matching the arguments.
@@ -1247,7 +1360,7 @@ class Bucket(_PropertyMixin):
         client = self._require_client(client)
         path = self.path + "/o"
         api_request = functools.partial(
-            client._connection.api_request, timeout=timeout, retry=DEFAULT_RETRY
+            client._connection.api_request, timeout=timeout, retry=retry
         )
         iterator = page_iterator.HTTPIterator(
             client=client,
@@ -1263,7 +1376,9 @@ class Bucket(_PropertyMixin):
         iterator.prefixes = set()
         return iterator
 
-    def list_notifications(self, client=None, timeout=_DEFAULT_TIMEOUT):
+    def list_notifications(
+        self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+    ):
         """List Pub / Sub notifications for this bucket.
 
         See:
@@ -1282,13 +1397,27 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: list of :class:`.BucketNotification`
         :returns: notification instances
         """
         client = self._require_client(client)
         path = self.path + "/notificationConfigs"
         api_request = functools.partial(
-            client._connection.api_request, timeout=timeout, retry=DEFAULT_RETRY
+            client._connection.api_request, timeout=timeout, retry=retry
         )
         iterator = page_iterator.HTTPIterator(
             client=client,
@@ -1299,7 +1428,13 @@ class Bucket(_PropertyMixin):
         iterator.bucket = self
         return iterator
 
-    def get_notification(self, notification_id, client=None, timeout=_DEFAULT_TIMEOUT):
+    def get_notification(
+        self,
+        notification_id,
+        client=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
+    ):
         """Get Pub / Sub notification for this bucket.
 
         See:
@@ -1321,6 +1456,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`.BucketNotification`
         :returns: notification instance.
 
@@ -1334,7 +1483,7 @@ class Bucket(_PropertyMixin):
 
         """
         notification = self.notification(notification_id=notification_id)
-        notification.reload(client=client, timeout=timeout)
+        notification.reload(client=client, timeout=timeout, retry=retry)
         return notification
 
     def delete(
@@ -1344,6 +1493,7 @@ class Bucket(_PropertyMixin):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY,
     ):
         """Delete this bucket.
 
@@ -1385,6 +1535,20 @@ class Bucket(_PropertyMixin):
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises: :class:`ValueError` if ``force`` is ``True`` and the bucket
                  contains more than 256 objects / blobs.
         """
@@ -1430,7 +1594,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
 
     def delete_blob(
@@ -1443,6 +1607,7 @@ class Bucket(_PropertyMixin):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Deletes a blob from the current bucket.
 
@@ -1498,6 +1663,20 @@ class Bucket(_PropertyMixin):
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises: :class:`google.cloud.exceptions.NotFound` (to suppress
                  the exception, call ``delete_blobs``, passing a no-op
                  ``on_error`` callback, e.g.:
@@ -1528,7 +1707,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            retry=retry,
         )
 
     def delete_blobs(
@@ -1541,6 +1720,7 @@ class Bucket(_PropertyMixin):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Deletes a list of blobs from the current bucket.
 
@@ -1595,6 +1775,20 @@ class Bucket(_PropertyMixin):
                                             blob's current metageneration does not match the given value.
                                             The list must match ``blobs`` item-to-item.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises: :class:`~google.cloud.exceptions.NotFound` (if
                  `on_error` is not passed).
 
@@ -1637,6 +1831,7 @@ class Bucket(_PropertyMixin):
                     if_generation_not_match=next(if_generation_not_match, None),
                     if_metageneration_match=next(if_metageneration_match, None),
                     if_metageneration_not_match=next(if_metageneration_not_match, None),
+                    retry=retry,
                 )
             except NotFound:
                 if on_error is not None:
@@ -1661,6 +1856,7 @@ class Bucket(_PropertyMixin):
         if_source_generation_not_match=None,
         if_source_metageneration_match=None,
         if_source_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Copy the given blob to the given bucket, optionally with a new name.
 
@@ -1754,6 +1950,20 @@ class Bucket(_PropertyMixin):
                                                    object's current metageneration
                                                    does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`google.cloud.storage.blob.Blob`
         :returns: The new Blob.
 
@@ -1803,7 +2013,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=new_blob,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+            retry=retry,
         )
 
         if not preserve_acl:
@@ -1826,6 +2036,7 @@ class Bucket(_PropertyMixin):
         if_source_generation_not_match=None,
         if_source_metageneration_match=None,
         if_source_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         """Rename the given blob using copy and delete operations.
 
@@ -1916,6 +2127,20 @@ class Bucket(_PropertyMixin):
                                                    object's current metageneration
                                                    does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`Blob`
         :returns: The newly-renamed blob.
         """
@@ -1935,6 +2160,7 @@ class Bucket(_PropertyMixin):
             if_source_generation_not_match=if_source_generation_not_match,
             if_source_metageneration_match=if_source_metageneration_match,
             if_source_metageneration_not_match=if_source_metageneration_not_match,
+            retry=retry,
         )
 
         if not same_name:
@@ -1945,6 +2171,7 @@ class Bucket(_PropertyMixin):
                 if_generation_not_match=if_generation_not_match,
                 if_metageneration_match=if_metageneration_match,
                 if_metageneration_not_match=if_metageneration_not_match,
+                retry=retry,
             )
         return new_blob
 
@@ -2577,7 +2804,11 @@ class Bucket(_PropertyMixin):
         return self.configure_website(None, None)
 
     def get_iam_policy(
-        self, client=None, requested_policy_version=None, timeout=_DEFAULT_TIMEOUT
+        self,
+        client=None,
+        requested_policy_version=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Retrieve the IAM policy for the bucket.
 
@@ -2610,6 +2841,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
                   the ``getIamPolicy`` API request.
@@ -2653,11 +2898,17 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
         return Policy.from_api_repr(info)
 
-    def set_iam_policy(self, policy, client=None, timeout=_DEFAULT_TIMEOUT):
+    def set_iam_policy(
+        self,
+        policy,
+        client=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY_IF_ETAG_IN_JSON,
+    ):
         """Update the IAM policy for the bucket.
 
         See
@@ -2680,6 +2931,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
                   the ``setIamPolicy`` API request.
@@ -2699,11 +2964,13 @@ class Bucket(_PropertyMixin):
             data=resource,
             _target_object=None,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_ETAG_IN_JSON,
+            retry=retry,
         )
         return Policy.from_api_repr(info)
 
-    def test_iam_permissions(self, permissions, client=None, timeout=_DEFAULT_TIMEOUT):
+    def test_iam_permissions(
+        self, permissions, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+    ):
         """API call:  test permissions
 
         See
@@ -2726,6 +2993,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: list of string
         :returns: the permissions returned by the ``testIamPermissions`` API
                   request.
@@ -2742,12 +3023,17 @@ class Bucket(_PropertyMixin):
             path=path,
             query_params=query_params,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
         return resp.get("permissions", [])
 
     def make_public(
-        self, recursive=False, future=False, client=None, timeout=_DEFAULT_TIMEOUT
+        self,
+        recursive=False,
+        future=False,
+        client=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Update bucket's ACL, granting read access to anonymous users.
 
@@ -2771,6 +3057,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises ValueError:
             If ``recursive`` is True, and the bucket contains more than 256
             blobs.  This is to prevent extremely long runtime of this
@@ -2796,6 +3096,7 @@ class Bucket(_PropertyMixin):
                     max_results=self._MAX_OBJECTS_FOR_ITERATION + 1,
                     client=client,
                     timeout=timeout,
+                    retry=retry,
                 )
             )
             if len(blobs) > self._MAX_OBJECTS_FOR_ITERATION:
@@ -2810,10 +3111,15 @@ class Bucket(_PropertyMixin):
 
             for blob in blobs:
                 blob.acl.all().grant_read()
-                blob.acl.save(client=client, timeout=timeout)
+                blob.acl.save(client=client, timeout=timeout, retry=retry)
 
     def make_private(
-        self, recursive=False, future=False, client=None, timeout=_DEFAULT_TIMEOUT
+        self,
+        recursive=False,
+        future=False,
+        client=None,
+        timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Update bucket's ACL, revoking read access for anonymous users.
 
@@ -2838,6 +3144,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises ValueError:
             If ``recursive`` is True, and the bucket contains more than 256
             blobs.  This is to prevent extremely long runtime of this
@@ -2863,6 +3183,7 @@ class Bucket(_PropertyMixin):
                     max_results=self._MAX_OBJECTS_FOR_ITERATION + 1,
                     client=client,
                     timeout=timeout,
+                    retry=retry,
                 )
             )
             if len(blobs) > self._MAX_OBJECTS_FOR_ITERATION:
@@ -2942,9 +3263,16 @@ class Bucket(_PropertyMixin):
 
         return fields
 
-    def lock_retention_policy(self, client=None, timeout=_DEFAULT_TIMEOUT):
+    def lock_retention_policy(
+        self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+    ):
         """Lock the bucket's retention policy.
 
+        :type client: :class:`~google.cloud.storage.client.Client` or
+                      ``NoneType``
+        :param client: (Optional) The client to use.  If not passed, falls back
+                       to the ``client`` stored on the blob's bucket.
+
         :type timeout: float or tuple
         :param timeout: (Optional) The amount of time, in seconds, to wait
             for the server response.
@@ -2952,6 +3280,20 @@ class Bucket(_PropertyMixin):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises ValueError:
             if the bucket has no metageneration (i.e., new or never reloaded);
             if the bucket has no retention policy assigned;
@@ -2982,7 +3324,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=self,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
         self._set_properties(api_response)
 
diff --git a/google/cloud/storage/client.py b/google/cloud/storage/client.py
index c211144..ab67cca 100644
--- a/google/cloud/storage/client.py
+++ b/google/cloud/storage/client.py
@@ -46,6 +46,7 @@ from google.cloud.storage.acl import BucketACL
 from google.cloud.storage.acl import DefaultObjectACL
 from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 from google.cloud.storage.retry import DEFAULT_RETRY
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
 
 
 _marker = object()
@@ -235,7 +236,9 @@ class Client(ClientWithProject):
         """
         return self._batch_stack.top
 
-    def get_service_account_email(self, project=None, timeout=_DEFAULT_TIMEOUT):
+    def get_service_account_email(
+        self, project=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY
+    ):
         """Get the email address of the project's GCS service account
 
         :type project: str
@@ -249,6 +252,20 @@ class Client(ClientWithProject):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: str
         :returns: service account email address
         """
@@ -256,7 +273,7 @@ class Client(ClientWithProject):
             project = self.project
         path = "/projects/%s/serviceAccount" % (project,)
         api_response = self._base_connection.api_request(
-            method="GET", path=path, timeout=timeout, retry=DEFAULT_RETRY,
+            method="GET", path=path, timeout=timeout, retry=retry,
         )
         return api_response["email_address"]
 
@@ -297,6 +314,7 @@ class Client(ClientWithProject):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """API call: retrieve a bucket via a GET request.
 
@@ -324,6 +342,20 @@ class Client(ClientWithProject):
                 Make the operation conditional on whether the blob's
                 current metageneration does not match the given value.
 
+            retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):
+                How to retry the RPC. A None value will disable retries.
+                A google.api_core.retry.Retry value will enable retries, and the object will
+                define retriable response codes and errors and configure backoff and timeout options.
+
+                A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+                activates it only if certain conditions are met. This class exists to provide safe defaults
+                for RPC calls that are not technically safe to retry normally (due to potential data
+                duplication or other side-effects) but become safe to retry if a condition such as
+                if_metageneration_match is set.
+
+                See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+                information on retry types and how to configure them.
+
         Returns:
             google.cloud.storage.bucket.Bucket
                 The bucket matching the name provided.
@@ -359,6 +391,7 @@ class Client(ClientWithProject):
             timeout=timeout,
             if_metageneration_match=if_metageneration_match,
             if_metageneration_not_match=if_metageneration_not_match,
+            retry=retry,
         )
         return bucket
 
@@ -368,6 +401,7 @@ class Client(ClientWithProject):
         timeout=_DEFAULT_TIMEOUT,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
     ):
         """Get a bucket by name, returning None if not found.
 
@@ -397,6 +431,20 @@ class Client(ClientWithProject):
         :param if_metageneration_not_match: (Optional) Make the operation conditional on whether the
                                             blob's current metageneration does not match the given value.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`google.cloud.storage.bucket.Bucket`
         :returns: The bucket matching the name provided or None if not found.
         """
@@ -406,6 +454,7 @@ class Client(ClientWithProject):
                 timeout=timeout,
                 if_metageneration_match=if_metageneration_match,
                 if_metageneration_not_match=if_metageneration_not_match,
+                retry=retry,
             )
         except NotFound:
             return None
@@ -420,6 +469,7 @@ class Client(ClientWithProject):
         predefined_acl=None,
         predefined_default_object_acl=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """API call: create a new bucket via a POST request.
 
@@ -458,6 +508,20 @@ class Client(ClientWithProject):
                 Can also be passed as a tuple (connect_timeout, read_timeout).
                 See :meth:`requests.Session.request` documentation for details.
 
+            retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):
+                How to retry the RPC. A None value will disable retries.
+                A google.api_core.retry.Retry value will enable retries, and the object will
+                define retriable response codes and errors and configure backoff and timeout options.
+
+                A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+                activates it only if certain conditions are met. This class exists to provide safe defaults
+                for RPC calls that are not technically safe to retry normally (due to potential data
+                duplication or other side-effects) but become safe to retry if a condition such as
+                if_metageneration_match is set.
+
+                See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+                information on retry types and how to configure them.
+
         Returns:
             google.cloud.storage.bucket.Bucket
                 The newly created bucket.
@@ -532,7 +596,7 @@ class Client(ClientWithProject):
             data=properties,
             _target_object=bucket,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
 
         bucket._set_properties(api_response)
@@ -598,6 +662,7 @@ class Client(ClientWithProject):
         projection="noAcl",
         fields=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -668,6 +733,20 @@ class Client(ClientWithProject):
                 Can also be passed as a tuple (connect_timeout, read_timeout).
                 See :meth:`requests.Session.request` documentation for details.
 
+            retry (Optional[Union[google.api_core.retry.Retry, google.cloud.storage.retry.ConditionalRetryPolicy]]):
+                How to retry the RPC. A None value will disable retries.
+                A google.api_core.retry.Retry value will enable retries, and the object will
+                define retriable response codes and errors and configure backoff and timeout options.
+
+                A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+                activates it only if certain conditions are met. This class exists to provide safe defaults
+                for RPC calls that are not technically safe to retry normally (due to potential data
+                duplication or other side-effects) but become safe to retry if a condition such as
+                if_metageneration_match is set.
+
+                See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+                information on retry types and how to configure them.
+
         Returns:
             Iterator of all :class:`~google.cloud.storage.blob.Blob`
             in this bucket matching the arguments.
@@ -695,6 +774,7 @@ class Client(ClientWithProject):
             fields=fields,
             client=self,
             timeout=timeout,
+            retry=retry,
         )
 
     def list_buckets(
@@ -706,6 +786,7 @@ class Client(ClientWithProject):
         fields=None,
         project=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """Get all buckets in the project associated to the client.
 
@@ -757,6 +838,20 @@ class Client(ClientWithProject):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :raises ValueError: if both ``project`` is ``None`` and the client's
                             project is also ``None``.
@@ -780,7 +875,7 @@ class Client(ClientWithProject):
             extra_params["fields"] = fields
 
         api_request = functools.partial(
-            self._connection.api_request, retry=DEFAULT_RETRY, timeout=timeout
+            self._connection.api_request, retry=retry, timeout=timeout
         )
 
         return page_iterator.HTTPIterator(
@@ -852,6 +947,7 @@ class Client(ClientWithProject):
         project_id=None,
         user_project=None,
         timeout=_DEFAULT_TIMEOUT,
+        retry=DEFAULT_RETRY,
     ):
         """List HMAC keys for a project.
 
@@ -882,6 +978,20 @@ class Client(ClientWithProject):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype:
             Tuple[:class:`~google.cloud.storage.hmac_key.HMACKeyMetadata`, str]
         :returns: metadata for the created key, plus the bytes of the key's secret, which is an 40-character base64-encoded string.
@@ -902,7 +1012,7 @@ class Client(ClientWithProject):
             extra_params["userProject"] = user_project
 
         api_request = functools.partial(
-            self._connection.api_request, timeout=timeout, retry=DEFAULT_RETRY
+            self._connection.api_request, timeout=timeout, retry=retry
         )
 
         return page_iterator.HTTPIterator(
diff --git a/google/cloud/storage/hmac_key.py b/google/cloud/storage/hmac_key.py
index 796aeee..47ca33c 100644
--- a/google/cloud/storage/hmac_key.py
+++ b/google/cloud/storage/hmac_key.py
@@ -189,7 +189,7 @@ class HMACKeyMetadata(object):
         """
         return self._user_project
 
-    def exists(self, timeout=_DEFAULT_TIMEOUT):
+    def exists(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY):
         """Determine whether or not the key for this metadata exists.
 
         :type timeout: float or tuple
@@ -199,6 +199,20 @@ class HMACKeyMetadata(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: bool
         :returns: True if the key exists in Cloud Storage.
         """
@@ -209,14 +223,18 @@ class HMACKeyMetadata(object):
                 qs_params["userProject"] = self.user_project
 
             self._client._connection.api_request(
-                method="GET", path=self.path, query_params=qs_params, timeout=timeout
+                method="GET",
+                path=self.path,
+                query_params=qs_params,
+                timeout=timeout,
+                retry=retry,
             )
         except NotFound:
             return False
         else:
             return True
 
-    def reload(self, timeout=_DEFAULT_TIMEOUT):
+    def reload(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY):
         """Reload properties from Cloud Storage.
 
         :type timeout: float or tuple
@@ -226,6 +244,20 @@ class HMACKeyMetadata(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises :class:`~google.api_core.exceptions.NotFound`:
             if the key does not exist on the back-end.
         """
@@ -235,10 +267,14 @@ class HMACKeyMetadata(object):
             qs_params["userProject"] = self.user_project
 
         self._properties = self._client._connection.api_request(
-            method="GET", path=self.path, query_params=qs_params, timeout=timeout
+            method="GET",
+            path=self.path,
+            query_params=qs_params,
+            timeout=timeout,
+            retry=retry,
         )
 
-    def update(self, timeout=_DEFAULT_TIMEOUT):
+    def update(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY_IF_ETAG_IN_JSON):
         """Save writable properties to Cloud Storage.
 
         :type timeout: float or tuple
@@ -248,6 +284,20 @@ class HMACKeyMetadata(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises :class:`~google.api_core.exceptions.NotFound`:
             if the key does not exist on the back-end.
         """
@@ -262,10 +312,10 @@ class HMACKeyMetadata(object):
             data=payload,
             query_params=qs_params,
             timeout=timeout,
-            retry=DEFAULT_RETRY_IF_ETAG_IN_JSON,
+            retry=retry,
         )
 
-    def delete(self, timeout=_DEFAULT_TIMEOUT):
+    def delete(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY):
         """Delete the key from Cloud Storage.
 
         :type timeout: float or tuple
@@ -275,6 +325,20 @@ class HMACKeyMetadata(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises :class:`~google.api_core.exceptions.NotFound`:
             if the key does not exist on the back-end.
         """
@@ -290,5 +354,5 @@ class HMACKeyMetadata(object):
             path=self.path,
             query_params=qs_params,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
diff --git a/google/cloud/storage/notification.py b/google/cloud/storage/notification.py
index 07333e6..fde5e45 100644
--- a/google/cloud/storage/notification.py
+++ b/google/cloud/storage/notification.py
@@ -275,7 +275,7 @@ class BucketNotification(object):
             retry=None,
         )
 
-    def exists(self, client=None, timeout=_DEFAULT_TIMEOUT):
+    def exists(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY):
         """Test whether this notification exists.
 
         See:
@@ -295,6 +295,20 @@ class BucketNotification(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :rtype: bool
         :returns: True, if the notification exists, else False.
         :raises ValueError: if the notification has no ID.
@@ -310,14 +324,18 @@ class BucketNotification(object):
 
         try:
             client._connection.api_request(
-                method="GET", path=self.path, query_params=query_params, timeout=timeout
+                method="GET",
+                path=self.path,
+                query_params=query_params,
+                timeout=timeout,
+                retry=retry,
             )
         except NotFound:
             return False
         else:
             return True
 
-    def reload(self, client=None, timeout=_DEFAULT_TIMEOUT):
+    def reload(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY):
         """Update this notification from the server configuration.
 
         See:
@@ -337,6 +355,21 @@ class BucketNotification(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
+
         :raises ValueError: if the notification has no ID.
         """
         if self.notification_id is None:
@@ -353,11 +386,11 @@ class BucketNotification(object):
             path=self.path,
             query_params=query_params,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
         self._set_properties(response)
 
-    def delete(self, client=None, timeout=_DEFAULT_TIMEOUT):
+    def delete(self, client=None, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY):
         """Delete this notification.
 
         See:
@@ -377,6 +410,20 @@ class BucketNotification(object):
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
+        :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy
+        :param retry: (Optional) How to retry the RPC. A None value will disable retries.
+            A google.api_core.retry.Retry value will enable retries, and the object will
+            define retriable response codes and errors and configure backoff and timeout options.
+
+            A google.cloud.storage.retry.ConditionalRetryPolicy value wraps a Retry object and
+            activates it only if certain conditions are met. This class exists to provide safe defaults
+            for RPC calls that are not technically safe to retry normally (due to potential data
+            duplication or other side-effects) but become safe to retry if a condition such as
+            if_metageneration_match is set.
+
+            See the retry.py source code and docstrings in this package (google.cloud.storage.retry) for
+            information on retry types and how to configure them.
+
         :raises: :class:`google.api_core.exceptions.NotFound`:
             if the notification does not exist.
         :raises ValueError: if the notification has no ID.
@@ -395,7 +442,7 @@ class BucketNotification(object):
             path=self.path,
             query_params=query_params,
             timeout=timeout,
-            retry=DEFAULT_RETRY,
+            retry=retry,
         )
 
 


Test patch:
diff --git a/tests/unit/test_blob.py b/tests/unit/test_blob.py
index fc6eda6..28f4e31 100644
--- a/tests/unit/test_blob.py
+++ b/tests/unit/test_blob.py
@@ -26,6 +26,7 @@ import pytest
 import six
 from six.moves import http_client
 
+from google.cloud.storage.retry import DEFAULT_RETRY
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
 
 
@@ -662,6 +663,7 @@ class Test_Blob(unittest.TestCase):
                 "query_params": {"fields": "name"},
                 "_target_object": None,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -684,6 +686,7 @@ class Test_Blob(unittest.TestCase):
                 "query_params": {"fields": "name", "userProject": USER_PROJECT},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -706,6 +709,7 @@ class Test_Blob(unittest.TestCase):
                 "query_params": {"fields": "name", "generation": GENERATION},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -739,6 +743,7 @@ class Test_Blob(unittest.TestCase):
                 },
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -764,6 +769,7 @@ class Test_Blob(unittest.TestCase):
                     None,
                     None,
                     None,
+                    DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
                 )
             ],
         )
@@ -780,7 +786,20 @@ class Test_Blob(unittest.TestCase):
         blob.delete(timeout=42)
         self.assertFalse(blob.exists())
         self.assertEqual(
-            bucket._deleted, [(BLOB_NAME, None, GENERATION, 42, None, None, None, None)]
+            bucket._deleted,
+            [
+                (
+                    BLOB_NAME,
+                    None,
+                    GENERATION,
+                    42,
+                    None,
+                    None,
+                    None,
+                    None,
+                    DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+                )
+            ],
         )
 
     def test_delete_w_generation_match(self):
@@ -796,7 +815,19 @@ class Test_Blob(unittest.TestCase):
         self.assertFalse(blob.exists())
         self.assertEqual(
             bucket._deleted,
-            [(BLOB_NAME, None, GENERATION, 42, GENERATION, None, None, None)],
+            [
+                (
+                    BLOB_NAME,
+                    None,
+                    GENERATION,
+                    42,
+                    GENERATION,
+                    None,
+                    None,
+                    None,
+                    DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
+                )
+            ],
         )
 
     def test__get_transport(self):
@@ -3070,6 +3101,7 @@ class Test_Blob(unittest.TestCase):
                 "query_params": {},
                 "_target_object": None,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY,
             },
         )
 
@@ -3106,6 +3138,7 @@ class Test_Blob(unittest.TestCase):
                 "query_params": {"optionsRequestedPolicyVersion": 3},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             },
         )
 
@@ -3147,6 +3180,7 @@ class Test_Blob(unittest.TestCase):
                 "query_params": {"userProject": USER_PROJECT},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             },
         )
 
@@ -4765,6 +4799,7 @@ class _Bucket(object):
         if_generation_not_match=None,
         if_metageneration_match=None,
         if_metageneration_not_match=None,
+        retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
     ):
         del self._blobs[blob_name]
         self._deleted.append(
@@ -4777,6 +4812,7 @@ class _Bucket(object):
                 if_generation_not_match,
                 if_metageneration_match,
                 if_metageneration_not_match,
+                retry,
             )
         )
 
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 668db2d..255953d 100644
--- a/tests/unit/test_bucket.py
+++ b/tests/unit/test_bucket.py
@@ -666,6 +666,7 @@ class Test_Bucket(unittest.TestCase):
             "query_params": {"fields": "name"},
             "_target_object": None,
             "timeout": 42,
+            "retry": DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -700,6 +701,7 @@ class Test_Bucket(unittest.TestCase):
             },
             "_target_object": None,
             "timeout": 42,
+            "retry": DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -727,6 +729,7 @@ class Test_Bucket(unittest.TestCase):
             "query_params": {"fields": "name", "userProject": USER_PROJECT},
             "_target_object": None,
             "timeout": self._get_default_timeout(),
+            "retry": DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -1615,6 +1618,7 @@ class Test_Bucket(unittest.TestCase):
             if_generation_not_match=None,
             if_metageneration_match=None,
             if_metageneration_not_match=None,
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
         )
 
     def test_rename_blob_with_generation_match(self):
@@ -1665,6 +1669,7 @@ class Test_Bucket(unittest.TestCase):
             if_generation_not_match=None,
             if_metageneration_match=None,
             if_metageneration_not_match=None,
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
         )
 
     def test_rename_blob_to_itself(self):
@@ -2744,9 +2749,9 @@ class Test_Bucket(unittest.TestCase):
             def grant_read(self):
                 self._granted = True
 
-            def save(self, client=None, timeout=None):
+            def save(self, client=None, timeout=None, retry=DEFAULT_RETRY):
                 _saved.append(
-                    (self._bucket, self._name, self._granted, client, timeout)
+                    (self._bucket, self._name, self._granted, client, timeout, retry)
                 )
 
         def item_to_blob(self, item):
@@ -2763,10 +2768,10 @@ class Test_Bucket(unittest.TestCase):
         bucket.default_object_acl.loaded = True
 
         with mock.patch("google.cloud.storage.bucket._item_to_blob", new=item_to_blob):
-            bucket.make_public(recursive=True, timeout=42)
+            bucket.make_public(recursive=True, timeout=42, retry=DEFAULT_RETRY)
         self.assertEqual(list(bucket.acl), permissive)
         self.assertEqual(list(bucket.default_object_acl), [])
-        self.assertEqual(_saved, [(bucket, BLOB_NAME, True, None, 42)])
+        self.assertEqual(_saved, [(bucket, BLOB_NAME, True, None, 42, DEFAULT_RETRY)])
         kw = connection._requested
         self.assertEqual(len(kw), 2)
         self.assertEqual(kw[0]["method"], "PATCH")
@@ -2886,9 +2891,9 @@ class Test_Bucket(unittest.TestCase):
             def revoke_read(self):
                 self._granted = False
 
-            def save(self, client=None, timeout=None):
+            def save(self, client=None, timeout=None, retry=DEFAULT_RETRY):
                 _saved.append(
-                    (self._bucket, self._name, self._granted, client, timeout)
+                    (self._bucket, self._name, self._granted, client, timeout, retry)
                 )
 
         def item_to_blob(self, item):
@@ -2905,10 +2910,10 @@ class Test_Bucket(unittest.TestCase):
         bucket.default_object_acl.loaded = True
 
         with mock.patch("google.cloud.storage.bucket._item_to_blob", new=item_to_blob):
-            bucket.make_private(recursive=True, timeout=42)
+            bucket.make_private(recursive=True, timeout=42, retry=DEFAULT_RETRY)
         self.assertEqual(list(bucket.acl), no_permissions)
         self.assertEqual(list(bucket.default_object_acl), [])
-        self.assertEqual(_saved, [(bucket, BLOB_NAME, False, None, 42)])
+        self.assertEqual(_saved, [(bucket, BLOB_NAME, False, None, 42, DEFAULT_RETRY)])
         kw = connection._requested
         self.assertEqual(len(kw), 2)
         self.assertEqual(kw[0]["method"], "PATCH")
diff --git a/tests/unit/test_hmac_key.py b/tests/unit/test_hmac_key.py
index d4ac933..5761f4a 100644
--- a/tests/unit/test_hmac_key.py
+++ b/tests/unit/test_hmac_key.py
@@ -238,6 +238,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "query_params": {},
             "timeout": 42,
+            "retry": DEFAULT_RETRY,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -266,6 +267,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "query_params": {"userProject": user_project},
             "timeout": self._get_default_timeout(),
+            "retry": DEFAULT_RETRY,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -290,6 +292,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "query_params": {},
             "timeout": 42,
+            "retry": DEFAULT_RETRY,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -320,6 +323,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "query_params": {"userProject": user_project},
             "timeout": self._get_default_timeout(),
+            "retry": DEFAULT_RETRY,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
diff --git a/tests/unit/test_notification.py b/tests/unit/test_notification.py
index e49e801..7ecabfa 100644
--- a/tests/unit/test_notification.py
+++ b/tests/unit/test_notification.py
@@ -347,7 +347,11 @@ class TestBucketNotification(unittest.TestCase):
         self.assertFalse(notification.exists(timeout=42))
 
         api_request.assert_called_once_with(
-            method="GET", path=self.NOTIFICATION_PATH, query_params={}, timeout=42
+            method="GET",
+            path=self.NOTIFICATION_PATH,
+            query_params={},
+            timeout=42,
+            retry=DEFAULT_RETRY,
         )
 
     def test_exists_hit(self):
@@ -371,6 +375,7 @@ class TestBucketNotification(unittest.TestCase):
             path=self.NOTIFICATION_PATH,
             query_params={"userProject": USER_PROJECT},
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     def test_reload_wo_notification_id(self):