diff --git a/storage/google/cloud/storage/_helpers.py b/storage/google/cloud/storage/_helpers.py
index 93848daa1cde..28dac9f45288 100644
--- a/storage/google/cloud/storage/_helpers.py
+++ b/storage/google/cloud/storage/_helpers.py
@@ -19,6 +19,7 @@
 
 import base64
 from hashlib import md5
+import functools
 
 
 def _validate_name(name):
@@ -263,3 +264,10 @@ def _base64_md5hash(buffer_object):
     _write_buffer_to_hash(buffer_object, hash_obj)
     digest_bytes = hash_obj.digest()
     return base64.b64encode(digest_bytes)
+
+
+def _call_api(fcn_call, retry, **kwargs):
+    call = functools.partial(fcn_call, **kwargs)
+    if retry:
+        call = retry(call)
+    return call()
diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py
index 9c89c52b9e24..46efb8101dc3 100644
--- a/storage/google/cloud/storage/blob.py
+++ b/storage/google/cloud/storage/blob.py
@@ -541,7 +541,14 @@ def _get_download_url(self):
         return _add_query_parameters(base_url, name_value_pairs)
 
     def _do_download(
-        self, transport, file_obj, download_url, headers, start=None, end=None
+        self,
+        transport,
+        file_obj,
+        download_url,
+        headers,
+        start=None,
+        end=None,
+        num_retries=None,
     ):
         """Perform a download without any error handling.
 
@@ -567,11 +574,18 @@ def _do_download(
 
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
+
+        :type num_retries: int
+        :param num_retries: Number of download retries.
         """
         if self.chunk_size is None:
             download = Download(
                 download_url, stream=file_obj, headers=headers, start=start, end=end
             )
+            if num_retries is not None:
+                download._retry_strategy = resumable_media.RetryStrategy(
+                    max_retries=num_retries
+                )
             download.consume(transport)
         else:
             download = ChunkedDownload(
@@ -582,11 +596,17 @@ def _do_download(
                 start=start if start else 0,
                 end=end,
             )
+            if num_retries is not None:
+                download._retry_strategy = resumable_media.RetryStrategy(
+                    max_retries=num_retries
+                )
 
             while not download.finished:
                 download.consume_next_chunk(transport)
 
-    def download_to_file(self, file_obj, client=None, start=None, end=None):
+    def download_to_file(
+        self, file_obj, client=None, start=None, end=None, num_retries=None
+    ):
         """Download the contents of this blob into a file-like object.
 
         .. note::
@@ -626,6 +646,9 @@ def download_to_file(self, file_obj, client=None, start=None, end=None):
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
 
+        :type num_retries: int
+        :param num_retries: Number of download retries.
+
         :raises: :class:`google.cloud.exceptions.NotFound`
         """
         download_url = self._get_download_url()
@@ -634,11 +657,15 @@ def download_to_file(self, file_obj, client=None, start=None, end=None):
 
         transport = self._get_transport(client)
         try:
-            self._do_download(transport, file_obj, download_url, headers, start, end)
+            self._do_download(
+                transport, file_obj, download_url, headers, start, end, num_retries
+            )
         except resumable_media.InvalidResponse as exc:
             _raise_from_invalid_response(exc)
 
-    def download_to_filename(self, filename, client=None, start=None, end=None):
+    def download_to_filename(
+        self, filename, client=None, start=None, end=None, num_retries=None
+    ):
         """Download the contents of this blob into a named file.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -658,11 +685,20 @@ def download_to_filename(self, filename, client=None, start=None, end=None):
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
 
+        :type num_retries: int
+        :param num_retries: Number of download retries.
+
         :raises: :class:`google.cloud.exceptions.NotFound`
         """
         try:
             with open(filename, "wb") as file_obj:
-                self.download_to_file(file_obj, client=client, start=start, end=end)
+                self.download_to_file(
+                    file_obj,
+                    client=client,
+                    start=start,
+                    end=end,
+                    num_retries=num_retries,
+                )
         except resumable_media.DataCorruption:
             # Delete the corrupt downloaded file.
             os.remove(filename)
@@ -673,7 +709,7 @@ def download_to_filename(self, filename, client=None, start=None, end=None):
             mtime = time.mktime(updated.timetuple())
             os.utime(file_obj.name, (mtime, mtime))
 
-    def download_as_string(self, client=None, start=None, end=None):
+    def download_as_string(self, client=None, start=None, end=None, num_retries=None):
         """Download the contents of this blob as a string.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -690,12 +726,17 @@ def download_as_string(self, client=None, start=None, end=None):
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
 
+        :type num_retries: int
+        :param num_retries: Number of download retries.
+
         :rtype: bytes
         :returns: The data stored in this blob.
         :raises: :class:`google.cloud.exceptions.NotFound`
         """
         string_buffer = BytesIO()
-        self.download_to_file(string_buffer, client=client, start=start, end=end)
+        self.download_to_file(
+            string_buffer, client=client, start=start, end=end, num_retries=num_retries
+        )
         return string_buffer.getvalue()
 
     def _get_content_type(self, content_type, filename=None):
@@ -1162,7 +1203,12 @@ def upload_from_file(
             _raise_from_invalid_response(exc)
 
     def upload_from_filename(
-        self, filename, content_type=None, client=None, predefined_acl=None
+        self,
+        filename,
+        content_type=None,
+        client=None,
+        predefined_acl=None,
+        num_retries=None,
     ):
         """Upload this blob's contents from the content of a named file.
 
@@ -1200,6 +1246,10 @@ def upload_from_filename(
 
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
+
+        :type num_retries: int
+        :param num_retries: Number of upload retries. (Deprecated: This
+                            argument will be removed in a future release.)
         """
         content_type = self._get_content_type(content_type, filename=filename)
 
@@ -1210,11 +1260,17 @@ def upload_from_filename(
                 content_type=content_type,
                 client=client,
                 size=total_bytes,
+                num_retries=num_retries,
                 predefined_acl=predefined_acl,
             )
 
     def upload_from_string(
-        self, data, content_type="text/plain", client=None, predefined_acl=None
+        self,
+        data,
+        content_type="text/plain",
+        client=None,
+        predefined_acl=None,
+        num_retries=None,
     ):
         """Upload contents of this blob from the provided string.
 
@@ -1247,6 +1303,10 @@ def upload_from_string(
 
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
+
+        :type num_retries: int
+        :param num_retries: Number of upload retries. (Deprecated: This
+                            argument will be removed in a future release.)
         """
         data = _to_bytes(data, encoding="utf-8")
         string_buffer = BytesIO(data)
@@ -1255,6 +1315,7 @@ def upload_from_string(
             size=len(data),
             content_type=content_type,
             client=client,
+            num_retries=num_retries,
             predefined_acl=predefined_acl,
         )
 
diff --git a/storage/google/cloud/storage/bucket.py b/storage/google/cloud/storage/bucket.py
index 0f30620dbed3..0ae3bca851f8 100644
--- a/storage/google/cloud/storage/bucket.py
+++ b/storage/google/cloud/storage/bucket.py
@@ -19,6 +19,7 @@
 import datetime
 import json
 import warnings
+import functools
 
 import six
 
@@ -27,6 +28,7 @@
 from google.cloud._helpers import _datetime_to_rfc3339
 from google.cloud._helpers import _NOW
 from google.cloud._helpers import _rfc3339_to_datetime
+from google.cloud.storage._helpers import _call_api
 from google.cloud.exceptions import NotFound
 from google.api_core.iam import Policy
 from google.cloud.storage import _signing
@@ -40,7 +42,7 @@
 from google.cloud.storage.blob import Blob
 from google.cloud.storage.notification import BucketNotification
 from google.cloud.storage.notification import NONE_PAYLOAD_FORMAT
-
+from google.cloud.storage.retry import DEFAULT_RETRY
 
 _LOCATION_SETTER_MESSAGE = (
     "Assignment to 'Bucket.location' is deprecated, as it is only "
@@ -566,7 +568,7 @@ def notification(
             payload_format=payload_format,
         )
 
-    def exists(self, client=None):
+    def exists(self, client=None, retry=DEFAULT_RETRY):
         """Determines whether or not this bucket exists.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -576,6 +578,9 @@ def exists(self, client=None):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :rtype: bool
         :returns: True if the bucket exists in Cloud Storage.
         """
@@ -590,7 +595,9 @@ def exists(self, client=None):
         try:
             # We intentionally pass `_target_object=None` since fields=name
             # would limit the local properties.
-            client._connection.api_request(
+            _call_api(
+                client._connection.api_request,
+                retry,
                 method="GET",
                 path=self.path,
                 query_params=query_params,
@@ -603,7 +610,7 @@ def exists(self, client=None):
         except NotFound:
             return False
 
-    def create(self, client=None, project=None, location=None):
+    def create(self, client=None, project=None, location=None, retry=DEFAULT_RETRY):
         """Creates current bucket.
 
         If the bucket already exists, will raise
@@ -630,6 +637,9 @@ def create(self, client=None, project=None, location=None):
         :param location: Optional. The location of the bucket. If not passed,
                          the default location, US, will be used. See
                          https://cloud.google.com/storage/docs/bucket-locations
+
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
         """
         if self.user_project is not None:
             raise ValueError("Cannot create bucket with 'user_project' set.")
@@ -649,7 +659,9 @@ def create(self, client=None, project=None, location=None):
         if location is not None:
             properties["location"] = location
 
-        api_response = client._connection.api_request(
+        api_response = _call_api(
+            client._connection.api_request,
+            retry,
             method="POST",
             path="/b",
             query_params=query_params,
@@ -775,6 +787,7 @@ def list_blobs(
         projection="noAcl",
         fields=None,
         client=None,
+        retry=DEFAULT_RETRY,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -826,6 +839,9 @@ def list_blobs(
         :param client: (Optional) The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :returns: Iterator of all :class:`~google.cloud.storage.blob.Blob`
                   in this bucket matching the arguments.
@@ -851,7 +867,9 @@ def list_blobs(
         path = self.path + "/o"
         iterator = page_iterator.HTTPIterator(
             client=client,
-            api_request=client._connection.api_request,
+            api_request=functools.partial(
+                _call_api, client._connection.api_request, retry
+            ),
             path=path,
             item_to_value=_item_to_blob,
             page_token=page_token,
@@ -863,7 +881,7 @@ def list_blobs(
         iterator.prefixes = set()
         return iterator
 
-    def list_notifications(self, client=None):
+    def list_notifications(self, client=None, retry=DEFAULT_RETRY):
         """List Pub / Sub notifications for this bucket.
 
         See:
@@ -876,6 +894,9 @@ def list_notifications(self, client=None):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :rtype: list of :class:`.BucketNotification`
         :returns: notification instances
         """
@@ -883,14 +904,16 @@ def list_notifications(self, client=None):
         path = self.path + "/notificationConfigs"
         iterator = page_iterator.HTTPIterator(
             client=client,
-            api_request=client._connection.api_request,
+            api_request=functools.partial(
+                _call_api, client._connection.api_request, retry
+            ),
             path=path,
             item_to_value=_item_to_notification,
         )
         iterator.bucket = self
         return iterator
 
-    def delete(self, force=False, client=None):
+    def delete(self, force=False, client=None, retry=DEFAULT_RETRY):
         """Delete this bucket.
 
         The bucket **must** be empty in order to submit a delete request. If
@@ -917,6 +940,9 @@ def delete(self, force=False, client=None):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :raises: :class:`ValueError` if ``force`` is ``True`` and the bucket
                  contains more than 256 objects / blobs.
         """
@@ -947,14 +973,16 @@ def delete(self, force=False, client=None):
         # We intentionally pass `_target_object=None` since a DELETE
         # request has no response value (whether in a standard request or
         # in a batch request).
-        client._connection.api_request(
+        _call_api(
+            client._connection.api_request,
+            retry,
             method="DELETE",
             path=self.path,
             query_params=query_params,
             _target_object=None,
         )
 
-    def delete_blob(self, blob_name, client=None, generation=None):
+    def delete_blob(self, blob_name, client=None, generation=None, retry=DEFAULT_RETRY):
         """Deletes a blob from the current bucket.
 
         If the blob isn't found (backend 404), raises a
@@ -980,6 +1008,9 @@ def delete_blob(self, blob_name, client=None, generation=None):
         :param generation: Optional. If present, permanently deletes a specific
                            revision of this object.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :raises: :class:`google.cloud.exceptions.NotFound` (to suppress
                  the exception, call ``delete_blobs``, passing a no-op
                  ``on_error`` callback, e.g.:
@@ -995,14 +1026,16 @@ def delete_blob(self, blob_name, client=None, generation=None):
         # We intentionally pass `_target_object=None` since a DELETE
         # request has no response value (whether in a standard request or
         # in a batch request).
-        client._connection.api_request(
+        _call_api(
+            client._connection.api_request,
+            retry,
             method="DELETE",
             path=blob.path,
             query_params=blob._query_params,
             _target_object=None,
         )
 
-    def delete_blobs(self, blobs, on_error=None, client=None):
+    def delete_blobs(self, blobs, on_error=None, client=None, retry=DEFAULT_RETRY):
         """Deletes a list of blobs from the current bucket.
 
         Uses :meth:`delete_blob` to delete each individual blob.
@@ -1023,6 +1056,9 @@ def delete_blobs(self, blobs, on_error=None, client=None):
         :param client: (Optional) The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :raises: :class:`~google.cloud.exceptions.NotFound` (if
                  `on_error` is not passed).
         """
@@ -1031,7 +1067,7 @@ def delete_blobs(self, blobs, on_error=None, client=None):
                 blob_name = blob
                 if not isinstance(blob_name, six.string_types):
                     blob_name = blob.name
-                self.delete_blob(blob_name, client=client)
+                self.delete_blob(blob_name, client=client, retry=retry)
             except NotFound:
                 if on_error is not None:
                     on_error(blob)
@@ -1046,6 +1082,7 @@ def copy_blob(
         client=None,
         preserve_acl=True,
         source_generation=None,
+        retry=DEFAULT_RETRY,
     ):
         """Copy the given blob to the given bucket, optionally with a new name.
 
@@ -1074,6 +1111,9 @@ def copy_blob(
         :param source_generation: Optional. The generation of the blob to be
                                   copied.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :rtype: :class:`google.cloud.storage.blob.Blob`
         :returns: The new Blob.
         """
@@ -1091,7 +1131,9 @@ def copy_blob(
 
         new_blob = Blob(bucket=destination_bucket, name=new_name)
         api_path = blob.path + "/copyTo" + new_blob.path
-        copy_result = client._connection.api_request(
+        copy_result = _call_api(
+            client._connection.api_request,
+            retry,
             method="POST",
             path=api_path,
             query_params=query_params,
@@ -1104,7 +1146,7 @@ def copy_blob(
         new_blob._set_properties(copy_result)
         return new_blob
 
-    def rename_blob(self, blob, new_name, client=None):
+    def rename_blob(self, blob, new_name, client=None, retry=DEFAULT_RETRY):
         """Rename the given blob using copy and delete operations.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -1129,12 +1171,15 @@ def rename_blob(self, blob, new_name, client=None):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :rtype: :class:`Blob`
         :returns: The newly-renamed blob.
         """
         same_name = blob.name == new_name
 
-        new_blob = self.copy_blob(blob, self, new_name, client=client)
+        new_blob = self.copy_blob(blob, self, new_name, client=client, retry=retry)
 
         if not same_name:
             blob.delete(client=client)
diff --git a/storage/google/cloud/storage/client.py b/storage/google/cloud/storage/client.py
index 511be69bc45f..3562302cf1d8 100644
--- a/storage/google/cloud/storage/client.py
+++ b/storage/google/cloud/storage/client.py
@@ -14,6 +14,8 @@
 
 """Client for interacting with the Google Cloud Storage API."""
 
+import functools
+
 from six.moves.urllib.parse import urlsplit
 
 from google.auth.credentials import AnonymousCredentials
@@ -26,6 +28,8 @@
 from google.cloud.storage.batch import Batch
 from google.cloud.storage.bucket import Bucket
 from google.cloud.storage.blob import Blob
+from google.cloud.storage.retry import DEFAULT_RETRY
+from google.cloud.storage._helpers import _call_api
 
 
 _marker = object()
@@ -298,7 +302,9 @@ def lookup_bucket(self, bucket_name):
         except NotFound:
             return None
 
-    def create_bucket(self, bucket_or_name, requester_pays=None, project=None):
+    def create_bucket(
+        self, bucket_or_name, requester_pays=None, project=None, retry=DEFAULT_RETRY
+    ):
         """API call: create a new bucket via a POST request.
 
         See
@@ -316,6 +322,8 @@ def create_bucket(self, bucket_or_name, requester_pays=None, project=None):
             project (str):
                 Optional. the project under which the bucket is to be created.
                 If not passed, uses the project set on the client.
+            retry (google.api_core.retry.Retry):
+                Optional. How to retry the API Call.
 
         Returns:
             google.cloud.storage.bucket.Bucket
@@ -350,10 +358,12 @@ def create_bucket(self, bucket_or_name, requester_pays=None, project=None):
 
         if requester_pays is not None:
             bucket.requester_pays = requester_pays
-        bucket.create(client=self, project=project)
+        bucket.create(client=self, project=project, retry=retry)
         return bucket
 
-    def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
+    def download_blob_to_file(
+        self, blob_or_uri, file_obj, start=None, end=None, num_retries=None
+    ):
         """Download the contents of a blob object or blob URI into a file-like object.
 
         Args:
@@ -368,6 +378,8 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
                 Optional. The first byte in a range to be downloaded.
             end (int):
                 Optional. The last byte in a range to be downloaded.
+            num_retries (int):
+                Optional. Number of download retries.
 
         Examples:
             Download a blob using using a blob resource.
@@ -394,7 +406,9 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
 
         """
         try:
-            blob_or_uri.download_to_file(file_obj, client=self, start=start, end=end)
+            blob_or_uri.download_to_file(
+                file_obj, client=self, start=start, end=end, num_retries=num_retries
+            )
         except AttributeError:
             scheme, netloc, path, query, frag = urlsplit(blob_or_uri)
             if scheme != "gs":
@@ -402,7 +416,9 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
             bucket = Bucket(self, name=netloc)
             blob_or_uri = Blob(path[1:], bucket)
 
-            blob_or_uri.download_to_file(file_obj, client=self, start=start, end=end)
+            blob_or_uri.download_to_file(
+                file_obj, client=self, start=start, end=end, num_retries=num_retries
+            )
 
     def list_blobs(
         self,
@@ -414,6 +430,7 @@ def list_blobs(
         versions=None,
         projection="noAcl",
         fields=None,
+        retry=DEFAULT_RETRY,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -462,6 +479,9 @@ def list_blobs(
                 ``'items(name,contentLanguage),nextPageToken'``.
                 See: https://cloud.google.com/storage/docs/json_api/v1/parameters#fields
 
+            retry (google.api_core.retry.Retry):
+                Optional. How to retry the API Call.
+
         Returns:
             Iterator of all :class:`~google.cloud.storage.blob.Blob`
             in this bucket matching the arguments.
@@ -476,6 +496,7 @@ def list_blobs(
             projection=projection,
             fields=fields,
             client=self,
+            retry=retry,
         )
 
     def list_buckets(
@@ -486,6 +507,7 @@ def list_buckets(
         projection="noAcl",
         fields=None,
         project=None,
+        retry=DEFAULT_RETRY,
     ):
         """Get all buckets in the project associated to the client.
 
@@ -532,6 +554,10 @@ def list_buckets(
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :raises ValueError: if both ``project`` is ``None`` and the client's
                             project is also ``None``.
+
+        :type retry: :class:`google.api_core.retry.Retry`
+        :param retry: (Optional) How to retry the API Call.
+
         :returns: Iterator of all :class:`~google.cloud.storage.bucket.Bucket`
                   belonging to this project.
         """
@@ -553,7 +579,9 @@ def list_buckets(
 
         return page_iterator.HTTPIterator(
             client=self,
-            api_request=self._connection.api_request,
+            api_request=functools.partial(
+                _call_api, self._connection.api_request, retry
+            ),
             path="/b",
             item_to_value=_item_to_bucket,
             page_token=page_token,
diff --git a/storage/google/cloud/storage/retry.py b/storage/google/cloud/storage/retry.py
new file mode 100644
index 000000000000..4bc4b757f45d
--- /dev/null
+++ b/storage/google/cloud/storage/retry.py
@@ -0,0 +1,55 @@
+# Copyright 2018 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.
+
+from google.api_core import exceptions
+from google.api_core import retry
+
+
+_RETRYABLE_REASONS = frozenset(
+    ["rateLimitExceeded", "backendError", "internalError", "badGateway"]
+)
+
+_UNSTRUCTURED_RETRYABLE_TYPES = (
+    exceptions.TooManyRequests,
+    exceptions.InternalServerError,
+    exceptions.BadGateway,
+)
+
+
+def _should_retry(exc):
+    """Predicate for determining when to retry.
+
+    We retry if and only if the 'reason' is 'backendError'
+    or 'rateLimitExceeded'.
+    """
+    if not hasattr(exc, "errors"):
+        return False
+
+    if len(exc.errors) == 0:
+        # Check for unstructured error returns, e.g. from GFE
+        return isinstance(exc, _UNSTRUCTURED_RETRYABLE_TYPES)
+
+    reason = exc.errors[0]["reason"]
+    return reason in _RETRYABLE_REASONS
+
+
+DEFAULT_RETRY = retry.Retry(predicate=_should_retry)
+"""The default retry object.
+
+Any method with a ``retry`` parameter will be retried automatically,
+with reasonable defaults. To disable retry, pass ``retry=None``.
+To modify the default retry behavior, call a ``with_XXX`` method
+on ``DEFAULT_RETRY``. For example, to change the deadline to 30 seconds,
+pass ``retry=bigquery.DEFAULT_RETRY.with_deadline(30)``.
+"""
diff --git a/storage/tests/unit/test_blob.py b/storage/tests/unit/test_blob.py
index b52a3753834a..57ac96a50ec7 100644
--- a/storage/tests/unit/test_blob.py
+++ b/storage/tests/unit/test_blob.py
@@ -807,7 +807,7 @@ def _check_session_mocks(self, client, transport, expected_url, headers=None):
         call = mock.call("GET", expected_url, data=None, headers=headers)
         self.assertEqual(transport.request.mock_calls, [call, call])
 
-    def test__do_download_simple(self):
+    def test__do_download_simple(self, num_retries=None):
         blob_name = "blob-name"
         # Create a fake client/bucket and use them in the Blob() constructor.
         client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
@@ -827,7 +827,9 @@ def test__do_download_simple(self):
         file_obj = io.BytesIO()
         download_url = "http://test.invalid"
         headers = {}
-        blob._do_download(transport, file_obj, download_url, headers)
+        blob._do_download(
+            transport, file_obj, download_url, headers, num_retries=num_retries
+        )
         # Make sure the download was as expected.
         self.assertEqual(file_obj.getvalue(), b"abcdef")
 
@@ -835,6 +837,9 @@ def test__do_download_simple(self):
             "GET", download_url, data=None, headers=headers, stream=True
         )
 
+    def test__test__do_download_simple_with_retry(self):
+        self.test__do_download_simple(num_retries=11)
+
     def test__do_download_simple_with_range(self):
         blob_name = "blob-name"
         # Create a fake client/bucket and use them in the Blob() constructor.
@@ -864,7 +869,7 @@ def test__do_download_simple_with_range(self):
             "GET", download_url, data=None, headers=headers, stream=True
         )
 
-    def test__do_download_chunked(self):
+    def test__do_download_chunked(self, num_retries=None):
         blob_name = "blob-name"
         # Create a fake client/bucket and use them in the Blob() constructor.
         client = mock.Mock(_credentials=_make_credentials(), spec=["_credentials"])
@@ -879,7 +884,9 @@ def test__do_download_chunked(self):
         file_obj = io.BytesIO()
         download_url = "http://test.invalid"
         headers = {}
-        blob._do_download(transport, file_obj, download_url, headers)
+        blob._do_download(
+            transport, file_obj, download_url, headers, num_retries=num_retries
+        )
         # Make sure the download was as expected.
         self.assertEqual(file_obj.getvalue(), b"abcdef")
 
@@ -890,6 +897,9 @@ def test__do_download_chunked(self):
         call = mock.call("GET", download_url, data=None, headers=headers)
         self.assertEqual(transport.request.mock_calls, [call, call])
 
+    def test__do_download_chunked_with_retry(self):
+        self.test__do_download_chunked(num_retries=11)
+
     def test__do_download_chunked_with_range(self):
         blob_name = "blob-name"
         # Create a fake client/bucket and use them in the Blob() constructor.
@@ -972,7 +982,7 @@ def test_download_to_file_wo_media_link(self):
         )
         self._check_session_mocks(client, transport, expected_url)
 
-    def _download_to_file_helper(self, use_chunks=False):
+    def _download_to_file_helper(self, use_chunks=False, num_retries=None):
         blob_name = "blob-name"
         transport = self._mock_download_transport()
         # Create a fake client/bucket and use them in the Blob() constructor.
@@ -996,7 +1006,7 @@ def _download_to_file_helper(self, use_chunks=False):
             transport.request.side_effect = [single_chunk_response]
 
         file_obj = io.BytesIO()
-        blob.download_to_file(file_obj)
+        blob.download_to_file(file_obj, num_retries=num_retries)
         self.assertEqual(file_obj.getvalue(), b"abcdef")
 
         if use_chunks:
@@ -1016,6 +1026,9 @@ def test_download_to_file_default(self):
     def test_download_to_file_with_chunk_size(self):
         self._download_to_file_helper(use_chunks=True)
 
+    def test_download_to_file_with_retry(self):
+        self._download_to_file_helper(num_retries=11)
+
     def _download_to_filename_helper(self, updated=None):
         import os
         import time
diff --git a/storage/tests/unit/test_bucket.py b/storage/tests/unit/test_bucket.py
index 9ac4995525cf..454253963cbc 100644
--- a/storage/tests/unit/test_bucket.py
+++ b/storage/tests/unit/test_bucket.py
@@ -14,8 +14,10 @@
 
 import datetime
 import unittest
-
+from six.moves import http_client
+import requests
 import mock
+import json
 
 
 def _create_signing_credentials():
@@ -31,6 +33,33 @@ class _SigningCredentials(
     return credentials
 
 
+def _make_response_error(
+    status=http_client.INTERNAL_SERVER_ERROR, content=b"", headers={}
+):
+    response = requests.Response()
+    response.status_code = status
+    response._content = content
+    response.headers = headers
+    response.request = requests.Request()
+    return response
+
+
+def _make_requests_session(responses):
+    session = mock.create_autospec(requests.Session, instance=True)
+    session.request.side_effect = responses
+    return session
+
+
+def _make_json_response_w_error(
+    data, status=http_client.INTERNAL_SERVER_ERROR, headers=None
+):
+    headers = headers or {}
+    headers["Content-Type"] = "application/json"
+    return _make_response_error(
+        status=status, content=json.dumps(data).encode("utf-8"), headers=headers
+    )
+
+
 class Test_LifecycleRuleConditions(unittest.TestCase):
     @staticmethod
     def _get_target_class():
@@ -569,6 +598,52 @@ def test_create_w_explicit_project(self):
         self.assertEqual(kw["query_params"], {"project": OTHER_PROJECT})
         self.assertEqual(kw["data"], DATA)
 
+    def test_create_bucket_retry_error(self):
+        from google.cloud.exceptions import ServerError
+
+        PROJECT = "PROJECT"
+        BUCKET_NAME = "bucket-name"
+        DATA = {"name": BUCKET_NAME}
+
+        # Retryable reason,default retry: success.
+        exc = ServerError("", errors=[{"reason": "backendError"}])
+        connection = _Connection(exc)
+        client = _Client(connection, project=PROJECT)
+        bucket = self._make_one(client=client, name=BUCKET_NAME)
+
+        http = _make_requests_session([_make_json_response_w_error(DATA)])
+        with mock.patch(
+            "google.cloud.storage._http.Connection.api_request",
+            new=mock.MagicMock(return_value=http),
+        ):
+            bucket.create()
+        self.assertEqual(bucket._properties, exc)
+        kw, = connection._requested
+        self.assertEqual(kw["method"], "POST")
+        self.assertEqual(kw["path"], "/b")
+        self.assertEqual(kw["query_params"], {"project": PROJECT})
+        self.assertEqual(kw["data"], DATA)
+
+    def test_create_bucket_retry_none(self):
+        from google.cloud.exceptions import ServerError
+
+        PROJECT = "PROJECT"
+        BUCKET_NAME = "bucket-name"
+
+        # Retryable reason, but retry is disabled.
+        exc = ServerError("", errors=[{"reason": "backendError"}])
+        connection = _Connection(exc)
+        client = _Client(connection, project=PROJECT)
+        bucket = self._make_one(client=client, name=BUCKET_NAME)
+
+        http = _make_requests_session([_make_json_response_w_error({})])
+        with mock.patch(
+            "google.cloud.storage._http.Connection.api_request",
+            new=mock.MagicMock(return_value=http),
+        ):
+            bucket.create(retry=None)
+        self.assertEqual(bucket._properties, exc)
+
     def test_create_w_explicit_location(self):
         PROJECT = "PROJECT"
         BUCKET_NAME = "bucket-name"
diff --git a/storage/tests/unit/test_client.py b/storage/tests/unit/test_client.py
index b5597977bb17..7e6d0bd8a6c3 100644
--- a/storage/tests/unit/test_client.py
+++ b/storage/tests/unit/test_client.py
@@ -619,7 +619,7 @@ def test_download_blob_to_file_with_blob(self):
 
         client.download_blob_to_file(blob, file_obj)
         blob.download_to_file.assert_called_once_with(
-            file_obj, client=client, start=None, end=None
+            file_obj, client=client, start=None, end=None, num_retries=None
         )
 
     def test_download_blob_to_file_with_uri(self):
@@ -633,7 +633,7 @@ def test_download_blob_to_file_with_uri(self):
             client.download_blob_to_file("gs://bucket_name/path/to/object", file_obj)
 
         blob.download_to_file.assert_called_once_with(
-            file_obj, client=client, start=None, end=None
+            file_obj, client=client, start=None, end=None, num_retries=None
         )
 
     def test_download_blob_to_file_with_invalid_uri(self):
@@ -822,6 +822,32 @@ def test_list_buckets_non_empty(self):
             method="GET", url=mock.ANY, data=mock.ANY, headers=mock.ANY
         )
 
+    def test_list_buckets_retry_none(self):
+        PROJECT = "foo-bar"
+        CREDENTIALS = _make_credentials()
+        client = self._make_one(project=PROJECT, credentials=CREDENTIALS)
+
+        MAX_RESULTS = 10
+        PAGE_TOKEN = "ABCD"
+        PREFIX = "subfolder"
+        PROJECTION = "full"
+        FIELDS = "items/id,nextPageToken"
+
+        data = {"items": []}
+        http = _make_requests_session([_make_json_response(data)])
+        client._http_internal = http
+
+        iterator = client.list_buckets(
+            max_results=MAX_RESULTS,
+            page_token=PAGE_TOKEN,
+            prefix=PREFIX,
+            projection=PROJECTION,
+            fields=FIELDS,
+            retry=None,
+        )
+        buckets = list(iterator)
+        self.assertEqual(buckets, [])
+
     def test_list_buckets_all_arguments(self):
         from six.moves.urllib.parse import parse_qs
         from six.moves.urllib.parse import urlparse
diff --git a/storage/tests/unit/test_retry.py b/storage/tests/unit/test_retry.py
new file mode 100644
index 000000000000..486ea0f1b78d
--- /dev/null
+++ b/storage/tests/unit/test_retry.py
@@ -0,0 +1,69 @@
+# Copyright 2018 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
+
+import mock
+
+
+class Test_should_retry(unittest.TestCase):
+    def _call_fut(self, exc):
+        from google.cloud.storage.retry import _should_retry
+
+        return _should_retry(exc)
+
+    def test_wo_errors_attribute(self):
+        self.assertFalse(self._call_fut(object()))
+
+    def test_w_empty_errors(self):
+        exc = mock.Mock(errors=[], spec=["errors"])
+        self.assertFalse(self._call_fut(exc))
+
+    def test_w_non_matching_reason(self):
+        exc = mock.Mock(errors=[{"reason": "bogus"}], spec=["errors"])
+        self.assertFalse(self._call_fut(exc))
+
+    def test_w_backendError(self):
+        exc = mock.Mock(errors=[{"reason": "backendError"}], spec=["errors"])
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_rateLimitExceeded(self):
+        exc = mock.Mock(errors=[{"reason": "rateLimitExceeded"}], spec=["errors"])
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_unstructured_too_many_requests(self):
+        from google.api_core.exceptions import TooManyRequests
+
+        exc = TooManyRequests("testing")
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_internalError(self):
+        exc = mock.Mock(errors=[{"reason": "internalError"}], spec=["errors"])
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_unstructured_internal_server_error(self):
+        from google.api_core.exceptions import InternalServerError
+
+        exc = InternalServerError("testing")
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_badGateway(self):
+        exc = mock.Mock(errors=[{"reason": "badGateway"}], spec=["errors"])
+        self.assertTrue(self._call_fut(exc))
+
+    def test_w_unstructured_bad_gateway(self):
+        from google.api_core.exceptions import BadGateway
+
+        exc = BadGateway("testing")
+        self.assertTrue(self._call_fut(exc))