Repository: googleapis/python-storage

Issue:
Add retry (exponential backoff) for storage api functions
While using the python-storage library, we were getting 503 or 504 errors from time to time when deleting or moving objects.
I looked up the [docs](https://cloud.google.com/storage/docs/exponential-backoff) for cloud storage and came across this:
> Clients should use truncated exponential backoff for all requests to Cloud Storage that return HTTP 5xx and 429 response codes, including uploads and downloads of data or metadata.

At the bottom of the page there is also:
> Examples of backoff implemented in client libraries that you can use with Cloud Storage include:
> - Retrying library for Python.
> - Google Cloud Client Libraries for Node.js can automatically use backoff strategies to retry requests with the autoRetry parameter.

It would be super useful and in my opinion necessary to include retrying functionality into the python-storage library. 
It was also a bit confusing to me that the official docs are recommending a third party retrying library while already having [google.api.core.retry](https://googleapis.dev/python/google-api-core/latest/retry.html) inside google cloud python sdk. 

The `google.api.core.retry` docs mention this:
> Some client library methods apply retry automatically. These methods can accept a retry parameter that allows you to configure the behavior

A potential solution could be to add the retry parameter of type `google.api.core.retry` to public api functions or while creating the storage client. Developers could specify if they want exponential backoff and set the parameters as they like (maximum delay, deadline, etc.)

Java and Node.js libraries already have similar functionality and it seems that this can be useful when cloud storage is facing some issues or high traffic load, or in the case where users are hitting the limits of 1k writes or 5k reads per second and GCS is autoscaling.

If this is already supported in the library, please direct me to some docs, I wasn't able to find it...

Discussion:
andrewsg: @tritone FYI
andrewsg: @crwilcox @frankyn PTAL. Changes include:

- Retry predicate simplified; my testing suggests the google-api-core nuanced inspection of error messages is not needed for GCS.
- Retry predicate now includes all HTTP error codes listed in retry design, except 408
-- 408 is mysteriously missing from google-api-core exceptions and I have no way to test what happens if we actually get one (testbench was no help here)
- Retry predicate now includes ConnectionError which encompasses failure to find host and broken connections of all types. LMK if we need to narrow it down; we have fairly rich and detailed info on the specific ConnectionError subtype which we can use here.

Responded to all other comments as well.
andrewsg: @frankyn Added documentation as requested, PTAL

Patch:
diff --git a/google/cloud/storage/_helpers.py b/google/cloud/storage/_helpers.py
index a1075ea..ba59f8f 100644
--- a/google/cloud/storage/_helpers.py
+++ b/google/cloud/storage/_helpers.py
@@ -24,6 +24,8 @@ import os
 
 from six.moves.urllib.parse import urlsplit
 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
 
 
 STORAGE_EMULATOR_ENV_VAR = "STORAGE_EMULATOR_HOST"
@@ -205,6 +207,7 @@ class _PropertyMixin(object):
             headers=self._encryption_headers(),
             _target_object=self,
             timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
         self._set_properties(api_response)
 
@@ -306,6 +309,7 @@ class _PropertyMixin(object):
             query_params=query_params,
             _target_object=self,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
         )
         self._set_properties(api_response)
 
@@ -368,6 +372,7 @@ class _PropertyMixin(object):
             if_metageneration_match=if_metageneration_match,
             if_metageneration_not_match=if_metageneration_not_match,
         )
+
         api_response = client._connection.api_request(
             method="PUT",
             path=self.path,
@@ -375,6 +380,7 @@ class _PropertyMixin(object):
             query_params=query_params,
             _target_object=self,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
         )
         self._set_properties(api_response)
 
diff --git a/google/cloud/storage/_http.py b/google/cloud/storage/_http.py
index 032f70e..6e17519 100644
--- a/google/cloud/storage/_http.py
+++ b/google/cloud/storage/_http.py
@@ -14,6 +14,8 @@
 
 """Create / interact with Google Cloud Storage connections."""
 
+import functools
+
 from google.cloud import _http
 
 from google.cloud.storage import __version__
@@ -46,3 +48,16 @@ class Connection(_http.JSONConnection):
 
     API_URL_TEMPLATE = "{api_base_url}/storage/{api_version}{path}"
     """A template for the URL of a particular API call."""
+
+    def api_request(self, *args, **kwargs):
+        retry = kwargs.pop("retry", None)
+        call = functools.partial(super(Connection, self).api_request, *args, **kwargs)
+        if retry:
+            # If this is a ConditionalRetryPolicy, check conditions.
+            try:
+                retry = retry.get_retry_policy_if_conditions_met(**kwargs)
+            except AttributeError:  # This is not a ConditionalRetryPolicy.
+                pass
+            if retry:
+                call = retry(call)
+        return call()
diff --git a/google/cloud/storage/blob.py b/google/cloud/storage/blob.py
index b1e1378..f63303a 100644
--- a/google/cloud/storage/blob.py
+++ b/google/cloud/storage/blob.py
@@ -74,6 +74,7 @@ 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_IF_GENERATION_SPECIFIED
 
 
 _API_ACCESS_ENDPOINT = "https://storage.googleapis.com"
@@ -2856,6 +2857,7 @@ class Blob(_PropertyMixin):
             data=request,
             _target_object=self,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
         )
         self._set_properties(api_response)
 
@@ -3000,6 +3002,7 @@ class Blob(_PropertyMixin):
             headers=headers,
             _target_object=self,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
         )
         rewritten = int(api_response["totalBytesRewritten"])
         size = int(api_response["objectSize"])
diff --git a/google/cloud/storage/bucket.py b/google/cloud/storage/bucket.py
index adf37d3..7ab9a13 100644
--- a/google/cloud/storage/bucket.py
+++ b/google/cloud/storage/bucket.py
@@ -57,7 +57,9 @@ from google.cloud.storage.constants import REGION_LOCATION_TYPE
 from google.cloud.storage.constants import STANDARD_STORAGE_CLASS
 from google.cloud.storage.notification import BucketNotification
 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
 
 _UBLA_BPO_ENABLED_MESSAGE = (
     "Pass only one of 'uniform_bucket_level_access_enabled' / "
@@ -1244,7 +1246,9 @@ class Bucket(_PropertyMixin):
 
         client = self._require_client(client)
         path = self.path + "/o"
-        api_request = functools.partial(client._connection.api_request, timeout=timeout)
+        api_request = functools.partial(
+            client._connection.api_request, timeout=timeout, retry=DEFAULT_RETRY
+        )
         iterator = page_iterator.HTTPIterator(
             client=client,
             api_request=api_request,
@@ -1283,7 +1287,9 @@ class Bucket(_PropertyMixin):
         """
         client = self._require_client(client)
         path = self.path + "/notificationConfigs"
-        api_request = functools.partial(client._connection.api_request, timeout=timeout)
+        api_request = functools.partial(
+            client._connection.api_request, timeout=timeout, retry=DEFAULT_RETRY
+        )
         iterator = page_iterator.HTTPIterator(
             client=client,
             api_request=api_request,
@@ -1424,6 +1430,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
 
     def delete_blob(
@@ -1521,6 +1528,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
         )
 
     def delete_blobs(
@@ -1795,6 +1803,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=new_blob,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
         )
 
         if not preserve_acl:
@@ -2644,6 +2653,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=None,
             timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
         return Policy.from_api_repr(info)
 
@@ -2689,6 +2699,7 @@ class Bucket(_PropertyMixin):
             data=resource,
             _target_object=None,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_ETAG_IN_JSON,
         )
         return Policy.from_api_repr(info)
 
@@ -2727,7 +2738,11 @@ class Bucket(_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=DEFAULT_RETRY,
         )
         return resp.get("permissions", [])
 
@@ -2967,6 +2982,7 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             _target_object=self,
             timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
         self._set_properties(api_response)
 
diff --git a/google/cloud/storage/client.py b/google/cloud/storage/client.py
index e522052..27c163a 100644
--- a/google/cloud/storage/client.py
+++ b/google/cloud/storage/client.py
@@ -45,6 +45,7 @@ from google.cloud.storage.hmac_key import HMACKeyMetadata
 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
 
 
 _marker = object()
@@ -255,7 +256,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
+            method="GET", path=path, timeout=timeout, retry=DEFAULT_RETRY,
         )
         return api_response["email_address"]
 
@@ -531,6 +532,7 @@ class Client(ClientWithProject):
             data=properties,
             _target_object=bucket,
             timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
 
         bucket._set_properties(api_response)
@@ -777,7 +779,9 @@ class Client(ClientWithProject):
         if fields is not None:
             extra_params["fields"] = fields
 
-        api_request = functools.partial(self._connection.api_request, timeout=timeout)
+        api_request = functools.partial(
+            self._connection.api_request, retry=DEFAULT_RETRY, timeout=timeout
+        )
 
         return page_iterator.HTTPIterator(
             client=self,
@@ -829,7 +833,11 @@ class Client(ClientWithProject):
             qs_params["userProject"] = user_project
 
         api_response = self._connection.api_request(
-            method="POST", path=path, query_params=qs_params, timeout=timeout
+            method="POST",
+            path=path,
+            query_params=qs_params,
+            timeout=timeout,
+            retry=None,
         )
         metadata = HMACKeyMetadata(self)
         metadata._properties = api_response["metadata"]
@@ -893,7 +901,9 @@ class Client(ClientWithProject):
         if user_project is not None:
             extra_params["userProject"] = user_project
 
-        api_request = functools.partial(self._connection.api_request, timeout=timeout)
+        api_request = functools.partial(
+            self._connection.api_request, timeout=timeout, retry=DEFAULT_RETRY
+        )
 
         return page_iterator.HTTPIterator(
             client=self,
diff --git a/google/cloud/storage/hmac_key.py b/google/cloud/storage/hmac_key.py
index d9c451c..796aeee 100644
--- a/google/cloud/storage/hmac_key.py
+++ b/google/cloud/storage/hmac_key.py
@@ -16,6 +16,8 @@ from google.cloud.exceptions import NotFound
 from google.cloud._helpers import _rfc3339_to_datetime
 
 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_ETAG_IN_JSON
 
 
 class HMACKeyMetadata(object):
@@ -260,6 +262,7 @@ class HMACKeyMetadata(object):
             data=payload,
             query_params=qs_params,
             timeout=timeout,
+            retry=DEFAULT_RETRY_IF_ETAG_IN_JSON,
         )
 
     def delete(self, timeout=_DEFAULT_TIMEOUT):
@@ -283,5 +286,9 @@ class HMACKeyMetadata(object):
             qs_params["userProject"] = self.user_project
 
         self._client._connection.api_request(
-            method="DELETE", path=self.path, query_params=qs_params, timeout=timeout
+            method="DELETE",
+            path=self.path,
+            query_params=qs_params,
+            timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
diff --git a/google/cloud/storage/notification.py b/google/cloud/storage/notification.py
index 434a44d..07333e6 100644
--- a/google/cloud/storage/notification.py
+++ b/google/cloud/storage/notification.py
@@ -19,6 +19,7 @@ import re
 from google.api_core.exceptions import NotFound
 
 from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+from google.cloud.storage.retry import DEFAULT_RETRY
 
 
 OBJECT_FINALIZE_EVENT_TYPE = "OBJECT_FINALIZE"
@@ -271,6 +272,7 @@ class BucketNotification(object):
             query_params=query_params,
             data=properties,
             timeout=timeout,
+            retry=None,
         )
 
     def exists(self, client=None, timeout=_DEFAULT_TIMEOUT):
@@ -347,7 +349,11 @@ class BucketNotification(object):
             query_params["userProject"] = self.bucket.user_project
 
         response = 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=DEFAULT_RETRY,
         )
         self._set_properties(response)
 
@@ -385,7 +391,11 @@ class BucketNotification(object):
             query_params["userProject"] = self.bucket.user_project
 
         client._connection.api_request(
-            method="DELETE", path=self.path, query_params=query_params, timeout=timeout
+            method="DELETE",
+            path=self.path,
+            query_params=query_params,
+            timeout=timeout,
+            retry=DEFAULT_RETRY,
         )
 
 
diff --git a/google/cloud/storage/retry.py b/google/cloud/storage/retry.py
new file mode 100644
index 0000000..c1f1ad1
--- /dev/null
+++ b/google/cloud/storage/retry.py
@@ -0,0 +1,136 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import requests
+
+from google.api_core import exceptions
+from google.api_core import retry
+
+import json
+
+
+_RETRYABLE_TYPES = (
+    exceptions.TooManyRequests,  # 429
+    exceptions.InternalServerError,  # 500
+    exceptions.BadGateway,  # 502
+    exceptions.ServiceUnavailable,  # 503
+    exceptions.GatewayTimeout,  # 504
+    requests.ConnectionError,
+)
+
+# Some retriable errors don't have their own custom exception in api_core.
+_ADDITIONAL_RETRYABLE_STATUS_CODES = (408,)
+
+
+def _should_retry(exc):
+    """Predicate for determining when to retry."""
+    if isinstance(exc, _RETRYABLE_TYPES):
+        return True
+    elif isinstance(exc, exceptions.GoogleAPICallError):
+        return exc.code in _ADDITIONAL_RETRYABLE_STATUS_CODES
+    else:
+        return False
+
+
+DEFAULT_RETRY = retry.Retry(predicate=_should_retry)
+"""The default retry object.
+
+This retry setting will retry all _RETRYABLE_TYPES and any status codes from
+_ADDITIONAL_RETRYABLE_STATUS_CODES.
+
+To modify the default retry behavior, create a new retry object modeled after
+this one by calling it a ``with_XXX`` method. For example, to create a copy of
+DEFAULT_RETRY with a deadline of 30 seconds, pass
+``retry=DEFAULT_RETRY.with_deadline(30)``. See google-api-core reference
+(https://googleapis.dev/python/google-api-core/latest/retry.html) for details.
+"""
+
+
+class ConditionalRetryPolicy(object):
+    """A class for use when an API call is only conditionally safe to retry.
+
+    This class is intended for use in inspecting the API call parameters of an
+    API call to verify that any flags necessary to make the API call idempotent
+    (such as specifying an ``if_generation_match`` or related flag) are present.
+
+    It can be used in place of a ``retry.Retry`` object, in which case
+    ``_http.Connection.api_request`` will pass the requested api call keyword
+    arguments into the ``conditional_predicate`` and return the ``retry_policy``
+    if the conditions are met.
+
+    :type retry_policy: class:`google.api_core.retry.Retry`
+    :param retry_policy: A retry object defining timeouts, persistence and which
+        exceptions to retry.
+
+    :type conditional_predicate: callable
+    :param conditional_predicate: A callable that accepts exactly the number of
+        arguments in ``required_kwargs``, in order, and returns True if the
+        arguments have sufficient data to determine that the call is safe to
+        retry (idempotent).
+
+    :type required_kwargs: list(str)
+    :param required_kwargs:
+        A list of keyword argument keys that will be extracted from the API call
+        and passed into the ``conditional predicate`` in order.
+    """
+
+    def __init__(self, retry_policy, conditional_predicate, required_kwargs):
+        self.retry_policy = retry_policy
+        self.conditional_predicate = conditional_predicate
+        self.required_kwargs = required_kwargs
+
+    def get_retry_policy_if_conditions_met(self, **kwargs):
+        if self.conditional_predicate(*[kwargs[key] for key in self.required_kwargs]):
+            return self.retry_policy
+        return None
+
+
+def is_generation_specified(query_params):
+    """Return True if generation or if_generation_match is specified."""
+    generation = query_params.get("generation") is not None
+    if_generation_match = query_params.get("if_generation_match") is not None
+    return generation or if_generation_match
+
+
+def is_metageneration_specified(query_params):
+    """Return True if if_metageneration_match is specified."""
+    if_metageneration_match = query_params.get("if_metageneration_match") is not None
+    return if_metageneration_match
+
+
+def is_etag_in_json(data):
+    """Return True if an etag is contained in the JSON body.
+
+    Indended for use on calls with relatively short JSON payloads."""
+    try:
+        content = json.loads(data)
+        if content.get("etag"):
+            return True
+    # Though this method should only be called when a JSON body is expected,
+    # the retry policy should be robust to unexpected payloads.
+    # In Python 3 a JSONDecodeError is possible, but it is a subclass of ValueError.
+    except (ValueError, TypeError):
+        pass
+    return False
+
+
+DEFAULT_RETRY_IF_GENERATION_SPECIFIED = ConditionalRetryPolicy(
+    DEFAULT_RETRY, is_generation_specified, ["query_params"]
+)
+DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED = ConditionalRetryPolicy(
+    DEFAULT_RETRY, is_metageneration_specified, ["query_params"]
+)
+DEFAULT_RETRY_IF_ETAG_IN_JSON = ConditionalRetryPolicy(
+    DEFAULT_RETRY, is_etag_in_json, ["data"]
+)


Test patch:
diff --git a/tests/unit/test__helpers.py b/tests/unit/test__helpers.py
index e295cbe..fa989f9 100644
--- a/tests/unit/test__helpers.py
+++ b/tests/unit/test__helpers.py
@@ -16,6 +16,9 @@ import unittest
 
 import mock
 
+from google.cloud.storage.retry import DEFAULT_RETRY
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+
 
 class Test__get_storage_host(unittest.TestCase):
     @staticmethod
@@ -122,6 +125,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "headers": {},
                 "_target_object": derived,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY,
             },
         )
         self.assertEqual(derived._changes, set())
@@ -158,6 +162,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "headers": {},
                 "_target_object": derived,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY,
             },
         )
         self.assertEqual(derived._changes, set())
@@ -183,6 +188,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "headers": {},
                 "_target_object": derived,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             },
         )
         self.assertEqual(derived._changes, set())
@@ -207,6 +213,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "headers": {},
                 "_target_object": derived,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY,
             },
         )
         self.assertEqual(derived._changes, set())
@@ -246,6 +253,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "data": {"bar": BAR},
                 "_target_object": derived,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
             },
         )
         # Make sure changes get reset by patch().
@@ -286,6 +294,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "data": {"bar": BAR},
                 "_target_object": derived,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
             },
         )
         # Make sure changes get reset by patch().
@@ -315,6 +324,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "data": {"bar": BAR},
                 "_target_object": derived,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED,
             },
         )
         # Make sure changes get reset by patch().
@@ -338,6 +348,7 @@ class Test_PropertyMixin(unittest.TestCase):
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
         self.assertEqual(kw[0]["data"], {"bar": BAR, "baz": BAZ})
         self.assertEqual(kw[0]["timeout"], 42)
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED)
         # Make sure changes get reset by patch().
         self.assertEqual(derived._changes, set())
 
@@ -366,6 +377,7 @@ class Test_PropertyMixin(unittest.TestCase):
         )
         self.assertEqual(kw[0]["data"], {"bar": BAR, "baz": BAZ})
         self.assertEqual(kw[0]["timeout"], 42)
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED)
         # Make sure changes get reset by patch().
         self.assertEqual(derived._changes, set())
 
@@ -390,6 +402,7 @@ class Test_PropertyMixin(unittest.TestCase):
         )
         self.assertEqual(kw[0]["data"], {"bar": BAR, "baz": BAZ})
         self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED)
         # Make sure changes get reset by patch().
         self.assertEqual(derived._changes, set())
 
diff --git a/tests/unit/test__http.py b/tests/unit/test__http.py
index 6ad1bf0..00cb4d3 100644
--- a/tests/unit/test__http.py
+++ b/tests/unit/test__http.py
@@ -99,3 +99,117 @@ class TestConnection(unittest.TestCase):
         self.assertEqual(path, "/".join(["", "storage", conn.API_VERSION, "foo"]))
         parms = dict(parse_qsl(qs))
         self.assertEqual(parms["bar"], "baz")
+
+    def test_api_request_no_retry(self):
+        import requests
+
+        http = mock.create_autospec(requests.Session, instance=True)
+        client = mock.Mock(_http=http, spec=["_http"])
+
+        conn = self._make_one(client)
+        response = requests.Response()
+        response.status_code = 200
+        data = b"brent-spiner"
+        response._content = data
+        http.request.return_value = response
+
+        req_data = "hey-yoooouuuuu-guuuuuyyssss"
+        conn.api_request("GET", "/rainbow", data=req_data, expect_json=False)
+        http.request.assert_called_once()
+
+    def test_api_request_basic_retry(self):
+        # For this test, the "retry" function will just short-circuit.
+        FAKE_RESPONSE_STRING = "fake_response"
+
+        def retry(_):
+            def fake_response():
+                return FAKE_RESPONSE_STRING
+
+            return fake_response
+
+        import requests
+
+        http = mock.create_autospec(requests.Session, instance=True)
+        client = mock.Mock(_http=http, spec=["_http"])
+
+        # Some of this is unnecessary if the test succeeds, but we'll leave it
+        # to ensure a failure produces a less confusing error message.
+        conn = self._make_one(client)
+        response = requests.Response()
+        response.status_code = 200
+        data = b"brent-spiner"
+        response._content = data
+        http.request.return_value = response
+
+        req_data = "hey-yoooouuuuu-guuuuuyyssss"
+        result = conn.api_request(
+            "GET", "/rainbow", data=req_data, expect_json=False, retry=retry
+        )
+        http.request.assert_not_called()
+        self.assertEqual(result, FAKE_RESPONSE_STRING)
+
+    def test_api_request_conditional_retry(self):
+        # For this test, the "retry" function will short-circuit.
+        FAKE_RESPONSE_STRING = "fake_response"
+
+        def retry(_):
+            def fake_response():
+                return FAKE_RESPONSE_STRING
+
+            return fake_response
+
+        conditional_retry_mock = mock.MagicMock()
+        conditional_retry_mock.get_retry_policy_if_conditions_met.return_value = retry
+
+        import requests
+
+        http = mock.create_autospec(requests.Session, instance=True)
+        client = mock.Mock(_http=http, spec=["_http"])
+
+        # Some of this is unnecessary if the test succeeds, but we'll leave it
+        # to ensure a failure produces a less confusing error message.
+        conn = self._make_one(client)
+        response = requests.Response()
+        response.status_code = 200
+        data = b"brent-spiner"
+        response._content = data
+        http.request.return_value = response
+
+        req_data = "hey-yoooouuuuu-guuuuuyyssss"
+        result = conn.api_request(
+            "GET",
+            "/rainbow",
+            data=req_data,
+            expect_json=False,
+            retry=conditional_retry_mock,
+        )
+        http.request.assert_not_called()
+        self.assertEqual(result, FAKE_RESPONSE_STRING)
+
+    def test_api_request_conditional_retry_failed(self):
+        conditional_retry_mock = mock.MagicMock()
+        conditional_retry_mock.get_retry_policy_if_conditions_met.return_value = None
+
+        import requests
+
+        http = mock.create_autospec(requests.Session, instance=True)
+        client = mock.Mock(_http=http, spec=["_http"])
+
+        # Some of this is unnecessary if the test succeeds, but we'll leave it
+        # to ensure a failure produces a less confusing error message.
+        conn = self._make_one(client)
+        response = requests.Response()
+        response.status_code = 200
+        data = b"brent-spiner"
+        response._content = data
+        http.request.return_value = response
+
+        req_data = "hey-yoooouuuuu-guuuuuyyssss"
+        conn.api_request(
+            "GET",
+            "/rainbow",
+            data=req_data,
+            expect_json=False,
+            retry=conditional_retry_mock,
+        )
+        http.request.assert_called_once()
diff --git a/tests/unit/test_blob.py b/tests/unit/test_blob.py
index f67b650..f713861 100644
--- a/tests/unit/test_blob.py
+++ b/tests/unit/test_blob.py
@@ -26,6 +26,8 @@ import pytest
 import six
 from six.moves import http_client
 
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+
 
 def _make_credentials():
     import google.auth.credentials
@@ -3218,6 +3220,7 @@ class Test_Blob(unittest.TestCase):
                 },
                 "_target_object": destination,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -3254,6 +3257,7 @@ class Test_Blob(unittest.TestCase):
                 },
                 "_target_object": destination,
                 "timeout": 42,
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -3295,6 +3299,7 @@ class Test_Blob(unittest.TestCase):
                 },
                 "_target_object": destination,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -3349,6 +3354,7 @@ class Test_Blob(unittest.TestCase):
                 },
                 "_target_object": destination,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
@@ -3418,6 +3424,7 @@ class Test_Blob(unittest.TestCase):
                 },
                 "_target_object": destination,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY_IF_GENERATION_SPECIFIED,
             },
         )
 
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 38a358d..668db2d 100644
--- a/tests/unit/test_bucket.py
+++ b/tests/unit/test_bucket.py
@@ -18,6 +18,10 @@ import unittest
 import mock
 import pytest
 
+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_METAGENERATION_SPECIFIED
+
 
 def _make_connection(*responses):
     import google.cloud.storage._http
@@ -1021,6 +1025,7 @@ class Test_Bucket(unittest.TestCase):
                 "query_params": {},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1042,6 +1047,7 @@ class Test_Bucket(unittest.TestCase):
                 "_target_object": None,
                 "query_params": {"userProject": USER_PROJECT},
                 "timeout": 42,
+                "retry": DEFAULT_RETRY,
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1065,6 +1071,7 @@ class Test_Bucket(unittest.TestCase):
                 "query_params": {},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1090,6 +1097,7 @@ class Test_Bucket(unittest.TestCase):
                 "query_params": {"ifMetagenerationMatch": METAGENERATION_NUMBER},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1112,6 +1120,7 @@ class Test_Bucket(unittest.TestCase):
                 "query_params": {},
                 "_target_object": None,
                 "timeout": self._get_default_timeout(),
+                "retry": DEFAULT_RETRY,
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1160,6 +1169,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw["query_params"], {"userProject": USER_PROJECT})
         self.assertEqual(kw["timeout"], 42)
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blob_hit_with_generation(self):
         NAME = "name"
@@ -1175,6 +1185,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw["query_params"], {"generation": GENERATION})
         self.assertEqual(kw["timeout"], self._get_default_timeout())
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blob_with_generation_match(self):
         NAME = "name"
@@ -1200,6 +1211,7 @@ class Test_Bucket(unittest.TestCase):
             {"ifGenerationMatch": GENERATION, "ifMetagenerationMatch": METAGENERATION},
         )
         self.assertEqual(kw["timeout"], self._get_default_timeout())
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blobs_empty(self):
         NAME = "name"
@@ -1223,6 +1235,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
         self.assertEqual(kw[0]["timeout"], 42)
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blobs_w_generation_match(self):
         NAME = "name"
@@ -1248,12 +1261,14 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(
             kw[0]["query_params"], {"ifGenerationMatch": GENERATION_NUMBER}
         )
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
         self.assertEqual(kw[1]["method"], "DELETE")
         self.assertEqual(kw[1]["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME2))
         self.assertEqual(kw[1]["timeout"], 42)
         self.assertEqual(
             kw[1]["query_params"], {"ifGenerationMatch": GENERATION_NUMBER2}
         )
+        self.assertEqual(kw[1]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blobs_w_generation_match_wrong_len(self):
         NAME = "name"
@@ -1295,10 +1310,12 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(
             kw[0]["query_params"], {"ifGenerationMatch": GENERATION_NUMBER}
         )
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
         self.assertEqual(kw[1]["method"], "DELETE")
         self.assertEqual(kw[1]["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME2))
         self.assertEqual(kw[1]["timeout"], 42)
         self.assertEqual(kw[1]["query_params"], {})
+        self.assertEqual(kw[1]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blobs_miss_no_on_error(self):
         from google.cloud.exceptions import NotFound
@@ -1315,9 +1332,11 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "DELETE")
         self.assertEqual(kw[0]["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
         self.assertEqual(kw[1]["method"], "DELETE")
         self.assertEqual(kw[1]["path"], "/b/%s/o/%s" % (NAME, NONESUCH))
         self.assertEqual(kw[1]["timeout"], self._get_default_timeout())
+        self.assertEqual(kw[1]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_delete_blobs_miss_w_on_error(self):
         NAME = "name"
@@ -1334,9 +1353,11 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "DELETE")
         self.assertEqual(kw[0]["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
+        self.assertEqual(kw[0]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
         self.assertEqual(kw[1]["method"], "DELETE")
         self.assertEqual(kw[1]["path"], "/b/%s/o/%s" % (NAME, NONESUCH))
         self.assertEqual(kw[1]["timeout"], self._get_default_timeout())
+        self.assertEqual(kw[1]["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_reload_bucket_w_metageneration_match(self):
         NAME = "name"
@@ -1385,6 +1406,7 @@ class Test_Bucket(unittest.TestCase):
             req["query_params"],
             {"projection": "full", "ifMetagenerationMatch": METAGENERATION_NUMBER},
         )
+        self.assertEqual(req["retry"], DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED)
 
     def test_update_bucket_w_generation_match(self):
         connection = _Connection({})
@@ -1426,6 +1448,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {})
         self.assertEqual(kw["timeout"], 42)
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_copy_blobs_source_generation(self):
         SOURCE = "source"
@@ -1452,6 +1475,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {"sourceGeneration": GENERATION})
         self.assertEqual(kw["timeout"], self._get_default_timeout())
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_copy_blobs_w_generation_match(self):
         SOURCE = "source"
@@ -1489,6 +1513,7 @@ class Test_Bucket(unittest.TestCase):
             },
         )
         self.assertEqual(kw["timeout"], self._get_default_timeout())
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_copy_blobs_preserve_acl(self):
         from google.cloud.storage.acl import ObjectACL
@@ -1522,6 +1547,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw1["path"], COPY_PATH)
         self.assertEqual(kw1["query_params"], {})
         self.assertEqual(kw1["timeout"], self._get_default_timeout())
+        self.assertEqual(kw1["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
         self.assertEqual(kw2["method"], "PATCH")
         self.assertEqual(kw2["path"], NEW_BLOB_PATH)
@@ -1553,6 +1579,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {"userProject": USER_PROJECT})
         self.assertEqual(kw["timeout"], self._get_default_timeout())
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
     def test_rename_blob(self):
         BUCKET_NAME = "BUCKET_NAME"
@@ -1579,6 +1606,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {})
         self.assertEqual(kw["timeout"], 42)
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
         blob.delete.assert_called_once_with(
             client=client,
@@ -1628,6 +1656,7 @@ class Test_Bucket(unittest.TestCase):
             },
         )
         self.assertEqual(kw["timeout"], 42)
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
         blob.delete.assert_called_once_with(
             client=client,
@@ -1660,6 +1689,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {})
         self.assertEqual(kw["timeout"], self._get_default_timeout())
+        self.assertEqual(kw["retry"], DEFAULT_RETRY_IF_GENERATION_SPECIFIED)
 
         blob.delete.assert_not_called()
 
@@ -2272,6 +2302,7 @@ class Test_Bucket(unittest.TestCase):
             data=DATA,
             _target_object=bucket,
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
         mock_warn.assert_called_with(
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index f987076..4efc35e 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -23,9 +23,14 @@ import unittest
 from six.moves import http_client
 from six.moves.urllib import parse as urlparse
 
+from google.api_core import exceptions
+
 from google.oauth2.service_account import Credentials
 from . import _read_local_json
 
+from google.cloud.storage.retry import DEFAULT_RETRY
+
+
 _SERVICE_ACCOUNT_JSON = _read_local_json("url_signer_v4_test_account.json")
 _CONFORMANCE_TESTS = _read_local_json("url_signer_v4_test_data.json")[
     "postPolicyV4Tests"
@@ -687,6 +692,7 @@ class TestClient(unittest.TestCase):
             data=data,
             _target_object=mock.ANY,
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     @mock.patch("warnings.warn")
@@ -759,6 +765,7 @@ class TestClient(unittest.TestCase):
             data=data,
             _target_object=bucket,
             timeout=42,
+            retry=DEFAULT_RETRY,
         )
 
     def test_create_bucket_w_predefined_default_object_acl_invalid(self):
@@ -794,6 +801,7 @@ class TestClient(unittest.TestCase):
             data=data,
             _target_object=bucket,
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     def test_create_bucket_w_explicit_location(self):
@@ -819,6 +827,7 @@ class TestClient(unittest.TestCase):
             _target_object=bucket,
             query_params={"project": project},
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
         self.assertEqual(bucket.location, location)
 
@@ -842,6 +851,7 @@ class TestClient(unittest.TestCase):
             data=DATA,
             _target_object=bucket,
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     def test_create_w_extra_properties(self):
@@ -893,6 +903,7 @@ class TestClient(unittest.TestCase):
             data=DATA,
             _target_object=bucket,
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     def test_create_hit(self):
@@ -914,6 +925,7 @@ class TestClient(unittest.TestCase):
             data=DATA,
             _target_object=bucket,
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     def test_create_bucket_w_string_success(self):
@@ -1057,6 +1069,7 @@ class TestClient(unittest.TestCase):
                 path="/b/%s/o" % BUCKET_NAME,
                 query_params={"projection": "noAcl"},
                 timeout=self._get_default_timeout(),
+                retry=DEFAULT_RETRY,
             )
 
     def test_list_blobs_w_all_arguments_and_user_project(self):
@@ -1121,6 +1134,7 @@ class TestClient(unittest.TestCase):
                 path="/b/%s/o" % BUCKET_NAME,
                 query_params=EXPECTED,
                 timeout=42,
+                retry=DEFAULT_RETRY,
             )
 
     def test_list_buckets_wo_project(self):
@@ -1827,6 +1841,33 @@ class TestClient(unittest.TestCase):
         self.assertEqual(fields["x-goog-signature"], EXPECTED_SIGN)
         self.assertEqual(fields["policy"], EXPECTED_POLICY)
 
+    def test_list_buckets_retries_error(self):
+        PROJECT = "PROJECT"
+        CREDENTIALS = _make_credentials()
+        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
+
+        BUCKET_NAME = "bucket-name"
+
+        data = {"items": [{"name": BUCKET_NAME}]}
+        http = _make_requests_session(
+            [exceptions.InternalServerError("mock error"), _make_json_response(data)]
+        )
+        client._http_internal = http
+
+        buckets = list(client.list_buckets())
+
+        self.assertEqual(len(buckets), 1)
+        self.assertEqual(buckets[0].name, BUCKET_NAME)
+
+        call = mock.call(
+            method="GET",
+            url=mock.ANY,
+            data=mock.ANY,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
+        )
+        http.request.assert_has_calls([call, call])
+
 
 @pytest.mark.parametrize("test_data", _POST_POLICY_TESTS)
 def test_conformance_post_policy(test_data):
diff --git a/tests/unit/test_hmac_key.py b/tests/unit/test_hmac_key.py
index a142939..d4ac933 100644
--- a/tests/unit/test_hmac_key.py
+++ b/tests/unit/test_hmac_key.py
@@ -16,6 +16,9 @@ import unittest
 
 import mock
 
+from google.cloud.storage.retry import DEFAULT_RETRY
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
+
 
 class TestHMACKeyMetadata(unittest.TestCase):
     @staticmethod
@@ -343,6 +346,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "data": {"state": "INACTIVE"},
             "query_params": {},
             "timeout": 42,
+            "retry": DEFAULT_RETRY_IF_ETAG_IN_JSON,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -376,6 +380,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "data": {"state": "ACTIVE"},
             "query_params": {"userProject": user_project},
             "timeout": self._get_default_timeout(),
+            "retry": DEFAULT_RETRY_IF_ETAG_IN_JSON,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -409,6 +414,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "query_params": {},
             "timeout": 42,
+            "retry": DEFAULT_RETRY,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -432,6 +438,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 f056701..e49e801 100644
--- a/tests/unit/test_notification.py
+++ b/tests/unit/test_notification.py
@@ -16,6 +16,8 @@ import unittest
 
 import mock
 
+from google.cloud.storage.retry import DEFAULT_RETRY
+
 
 class TestBucketNotification(unittest.TestCase):
 
@@ -269,6 +271,7 @@ class TestBucketNotification(unittest.TestCase):
             query_params={},
             data=data,
             timeout=self._get_default_timeout(),
+            retry=None,
         )
 
     def test_create_w_explicit_client(self):
@@ -320,6 +323,7 @@ class TestBucketNotification(unittest.TestCase):
             query_params={"userProject": USER_PROJECT},
             data=data,
             timeout=42,
+            retry=None,
         )
 
     def test_exists_wo_notification_id(self):
@@ -391,7 +395,11 @@ class TestBucketNotification(unittest.TestCase):
             notification.reload(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_reload_hit(self):
@@ -425,6 +433,7 @@ class TestBucketNotification(unittest.TestCase):
             path=self.NOTIFICATION_PATH,
             query_params={"userProject": USER_PROJECT},
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
     def test_delete_wo_notification_id(self):
@@ -449,7 +458,11 @@ class TestBucketNotification(unittest.TestCase):
             notification.delete(timeout=42)
 
         api_request.assert_called_once_with(
-            method="DELETE", path=self.NOTIFICATION_PATH, query_params={}, timeout=42
+            method="DELETE",
+            path=self.NOTIFICATION_PATH,
+            query_params={},
+            timeout=42,
+            retry=DEFAULT_RETRY,
         )
 
     def test_delete_hit(self):
@@ -468,6 +481,7 @@ class TestBucketNotification(unittest.TestCase):
             path=self.NOTIFICATION_PATH,
             query_params={"userProject": USER_PROJECT},
             timeout=self._get_default_timeout(),
+            retry=DEFAULT_RETRY,
         )
 
 
diff --git a/tests/unit/test_retry.py b/tests/unit/test_retry.py
new file mode 100644
index 0000000..7c5a6ba
--- /dev/null
+++ b/tests/unit/test_retry.py
@@ -0,0 +1,78 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+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_METAGENERATION_SPECIFIED
+from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
+
+
+class TestConditionalRetryPolicy(unittest.TestCase):
+    def test_is_generation_specified_match_metageneration(self):
+        conditional_policy = DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_generation_match": 1}
+        )
+        self.assertEqual(policy, DEFAULT_RETRY)
+
+    def test_is_generation_specified_match_generation(self):
+        conditional_policy = DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"generation": 1}
+        )
+        self.assertEqual(policy, DEFAULT_RETRY)
+
+    def test_is_generation_specified_mismatch(self):
+        conditional_policy = DEFAULT_RETRY_IF_GENERATION_SPECIFIED
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_metageneration_match": 1}
+        )
+        self.assertEqual(policy, None)
+
+    def test_is_metageneration_specified_match(self):
+        conditional_policy = DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_metageneration_match": 1}
+        )
+        self.assertEqual(policy, DEFAULT_RETRY)
+
+    def test_is_metageneration_specified_mismatch(self):
+        conditional_policy = DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_generation_match": 1}
+        )
+        self.assertEqual(policy, None)
+
+    def test_is_etag_in_json_etag_match(self):
+        conditional_policy = DEFAULT_RETRY_IF_ETAG_IN_JSON
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_generation_match": 1}, data='{"etag": "12345678"}'
+        )
+        self.assertEqual(policy, DEFAULT_RETRY)
+
+    def test_is_etag_in_json_mismatch(self):
+        conditional_policy = DEFAULT_RETRY_IF_ETAG_IN_JSON
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_generation_match": 1}, data="{}"
+        )
+        self.assertEqual(policy, None)
+
+    def test_is_meta_or_etag_in_json_invalid(self):
+        conditional_policy = DEFAULT_RETRY_IF_ETAG_IN_JSON
+        policy = conditional_policy.get_retry_policy_if_conditions_met(
+            query_params={"if_generation_match": 1}, data="I am invalid JSON!"
+        )
+        self.assertEqual(policy, None)