Repository: googleapis/python-storage

Issue:
Storage: add timeout parameter to all public methods
As a  library user, I would like to have a way to specify a (transport) timeout when calling methods that make HTTP requests under the hood. The timeout should have a reasonable default to prevent requests for hanging indefinitely in case I forget to pass in a `timeout` argument myself.

Motivation: User reports of requests hanging indefinitely, e.g. googleapis/google-cloud-python#10182.

Storage: No timeouts cause indefinite hanging
Library: Google Cloud Storage
Environment: Win7 and OSX
Python Version: 3.7.3
Google Cloud Storage Version: 1.25.0

I don't believe all methods of the storage client are using timeouts.  I've come across several situations where an upload or download has completely hung because of this.  Unfortunately, there's no stack trace because the thread is just hanging waiting for a response.  Just from a brief code inspection, I can identify an example area where a timeout is not being honored:

`Bucket.get_blob` calls `blob.reload()`, which then calls the following without specifying a timeout:
```
api_response = client._connection.api_request(
            method="GET",
            path=self.path,
            query_params=query_params,
            headers=self._encryption_headers(),
            _target_object=self,
        )
```
This then calls `JSONConnection.api_request` (defaults timeout to None) -> `JSONConnection._make_request` (defaults timeout to None) -> `JSONConnection._do_request` (defaults timeout to None) -> `AuthorizedSession.request` (defaults timeout to None), which makes the ultimate call to the `requests.Session` object with a `None` timeout.  The end result is that a request is made without a timeout, which can very easily cause a thread to hang indefinitely waiting for a response.

I realize that it would be a huge pain to try and find all possible `None` timeout paths and patch them, but I at least wanted to bring this to attention.  I'm currently wrapping every call to the Google Cloud Python library with a custom timeout function that forcefully stops execution after my own specified timeout, since I have no way to pass that in to the library.  A fix that either allows developers to pass in their custom timeout, either to each function called (e.g. `get_blob(...)`) or to the `client` object so that it's passed with every single request in the underlying `http` instance, would be amazing.  (In this sense, I suppose this issue is a mix of a bug and a feature request, so my apologies if I chose the incorrect category.)

Storage code for resumable uploads that makes the call to `resumable_media/requests/_helpers.py`, more specifically the `http_request` function, seems to do much better since that function sets a default timeout of (60, 61) as opposed to `None`.

Discussion:
plamut: System tests CI failure seems to be temporary (503 ServiceUnavailable).

Patch:
diff --git a/google/cloud/storage/_helpers.py b/google/cloud/storage/_helpers.py
index 5bfa13e..ae53a7b 100644
--- a/google/cloud/storage/_helpers.py
+++ b/google/cloud/storage/_helpers.py
@@ -21,6 +21,9 @@ import base64
 from hashlib import md5
 import os
 
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+
 STORAGE_EMULATOR_ENV_VAR = "STORAGE_EMULATOR_HOST"
 """Environment variable defining host for Storage emulator."""
 
@@ -117,7 +120,7 @@ class _PropertyMixin(object):
             params["userProject"] = self.user_project
         return params
 
-    def reload(self, client=None):
+    def reload(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Reload properties from Cloud Storage.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -126,6 +129,12 @@ class _PropertyMixin(object):
                       ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current object.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         client = self._require_client(client)
         query_params = self._query_params
@@ -138,6 +147,7 @@ class _PropertyMixin(object):
             query_params=query_params,
             headers=self._encryption_headers(),
             _target_object=self,
+            timeout=timeout,
         )
         self._set_properties(api_response)
 
@@ -169,7 +179,7 @@ class _PropertyMixin(object):
         # If the values are reset, the changes must as well.
         self._changes = set()
 
-    def patch(self, client=None):
+    def patch(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Sends all changed properties in a PATCH request.
 
         Updates the ``_properties`` with the response from the backend.
@@ -180,6 +190,12 @@ class _PropertyMixin(object):
                       ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current object.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         client = self._require_client(client)
         query_params = self._query_params
@@ -195,10 +211,11 @@ class _PropertyMixin(object):
             data=update_properties,
             query_params=query_params,
             _target_object=self,
+            timeout=timeout,
         )
         self._set_properties(api_response)
 
-    def update(self, client=None):
+    def update(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Sends all properties in a PUT request.
 
         Updates the ``_properties`` with the response from the backend.
@@ -209,6 +226,12 @@ class _PropertyMixin(object):
                       ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current object.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         client = self._require_client(client)
         query_params = self._query_params
@@ -219,6 +242,7 @@ class _PropertyMixin(object):
             data=self._properties,
             query_params=query_params,
             _target_object=self,
+            timeout=timeout,
         )
         self._set_properties(api_response)
 
diff --git a/google/cloud/storage/acl.py b/google/cloud/storage/acl.py
index 9b1af1d..7260c50 100644
--- a/google/cloud/storage/acl.py
+++ b/google/cloud/storage/acl.py
@@ -79,6 +79,8 @@ This list of tuples can be used as the ``entity`` and ``role`` fields
 when sending metadata for ACLs to the API.
 """
 
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
 
 class _ACLEntity(object):
     """Class representing a set of roles for an entity.
@@ -206,10 +208,18 @@ class ACL(object):
     def __init__(self):
         self.entities = {}
 
-    def _ensure_loaded(self):
-        """Load if not already loaded."""
+    def _ensure_loaded(self, timeout=_DEFAULT_TIMEOUT):
+        """Load if not already loaded.
+
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+        """
         if not self.loaded:
-            self.reload()
+            self.reload(timeout=timeout)
 
     @classmethod
     def validate_predefined(cls, predefined):
@@ -415,7 +425,7 @@ class ACL(object):
             client = self.client
         return client
 
-    def reload(self, client=None):
+    def reload(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Reload the ACL data from Cloud Storage.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -424,6 +434,12 @@ class ACL(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the ACL's parent.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         path = self.reload_path
         client = self._require_client(client)
@@ -435,13 +451,13 @@ class ACL(object):
         self.entities.clear()
 
         found = client._connection.api_request(
-            method="GET", path=path, query_params=query_params
+            method="GET", path=path, query_params=query_params, timeout=timeout
         )
         self.loaded = True
         for entry in found.get("items", ()):
             self.add_entity(self.entity_from_dict(entry))
 
-    def _save(self, acl, predefined, client):
+    def _save(self, acl, predefined, client, timeout=_DEFAULT_TIMEOUT):
         """Helper for :meth:`save` and :meth:`save_predefined`.
 
         :type acl: :class:`google.cloud.storage.acl.ACL`, or a compatible list.
@@ -457,6 +473,12 @@ class ACL(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the ACL's parent.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         query_params = {"projection": "full"}
         if predefined is not None:
@@ -474,13 +496,14 @@ class ACL(object):
             path=path,
             data={self._URL_PATH_ELEM: list(acl)},
             query_params=query_params,
+            timeout=timeout,
         )
         self.entities.clear()
         for entry in result.get(self._URL_PATH_ELEM, ()):
             self.add_entity(self.entity_from_dict(entry))
         self.loaded = True
 
-    def save(self, acl=None, client=None):
+    def save(self, acl=None, client=None, timeout=_DEFAULT_TIMEOUT):
         """Save this ACL for the current bucket.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -493,6 +516,12 @@ class ACL(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the ACL's parent.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         if acl is None:
             acl = self
@@ -501,9 +530,9 @@ class ACL(object):
             save_to_backend = True
 
         if save_to_backend:
-            self._save(acl, None, client)
+            self._save(acl, None, client, timeout=timeout)
 
-    def save_predefined(self, predefined, client=None):
+    def save_predefined(self, predefined, client=None, timeout=_DEFAULT_TIMEOUT):
         """Save this ACL for the current bucket using a predefined ACL.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -519,11 +548,17 @@ class ACL(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the ACL's parent.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         predefined = self.validate_predefined(predefined)
-        self._save(None, predefined, client)
+        self._save(None, predefined, client, timeout=timeout)
 
-    def clear(self, client=None):
+    def clear(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Remove all ACL entries.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -537,8 +572,14 @@ class ACL(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the ACL's parent.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
-        self.save([], client=client)
+        self.save([], client=client, timeout=timeout)
 
 
 class BucketACL(ACL):
diff --git a/google/cloud/storage/batch.py b/google/cloud/storage/batch.py
index 92f1a18..89425f9 100644
--- a/google/cloud/storage/batch.py
+++ b/google/cloud/storage/batch.py
@@ -29,6 +29,7 @@ import six
 from google.cloud import _helpers
 from google.cloud import exceptions
 from google.cloud.storage._http import Connection
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 
 
 class MIMEApplicationHTTP(MIMEApplication):
@@ -150,7 +151,9 @@ class Batch(Connection):
         self._requests = []
         self._target_objects = []
 
-    def _do_request(self, method, url, headers, data, target_object, timeout=None):
+    def _do_request(
+        self, method, url, headers, data, target_object, timeout=_DEFAULT_TIMEOUT
+    ):
         """Override Connection:  defer actual HTTP request.
 
         Only allow up to ``_MAX_BATCH_SIZE`` requests to be deferred.
@@ -175,7 +178,8 @@ class Batch(Connection):
 
         :type timeout: float or tuple
         :param timeout: (optional) The amount of time, in seconds, to wait
-            for the server response. By default, the method waits indefinitely.
+            for the server response.
+
             Can also be passed as a tuple (connect_timeout, read_timeout).
             See :meth:`requests.Session.request` documentation for details.
 
@@ -206,8 +210,8 @@ class Batch(Connection):
 
         multi = MIMEMultipart()
 
-        # Use timeout of last request, default to None (indefinite)
-        timeout = None
+        # Use timeout of last request, default to _DEFAULT_TIMEOUT
+        timeout = _DEFAULT_TIMEOUT
         for method, uri, headers, body, _timeout in self._requests:
             subrequest = MIMEApplicationHTTP(method, uri, headers, body)
             multi.attach(subrequest)
diff --git a/google/cloud/storage/blob.py b/google/cloud/storage/blob.py
index f134c3e..93beced 100644
--- a/google/cloud/storage/blob.py
+++ b/google/cloud/storage/blob.py
@@ -61,6 +61,7 @@ from google.cloud.storage._signing import generate_signed_url_v2
 from google.cloud.storage._signing import generate_signed_url_v4
 from google.cloud.storage.acl import ACL
 from google.cloud.storage.acl import ObjectACL
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 from google.cloud.storage.constants import ARCHIVE_STORAGE_CLASS
 from google.cloud.storage.constants import COLDLINE_STORAGE_CLASS
 from google.cloud.storage.constants import MULTI_REGIONAL_LEGACY_STORAGE_CLASS
@@ -509,7 +510,7 @@ class Blob(_PropertyMixin):
             access_token=access_token,
         )
 
-    def exists(self, client=None):
+    def exists(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Determines whether or not this blob exists.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -519,6 +520,12 @@ class Blob(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the blob's bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: bool
         :returns: True if the blob exists in Cloud Storage.
@@ -537,6 +544,7 @@ class Blob(_PropertyMixin):
                 path=self.path,
                 query_params=query_params,
                 _target_object=None,
+                timeout=timeout,
             )
             # NOTE: This will not fail immediately in a batch. However, when
             #       Batch.finish() is called, the resulting `NotFound` will be
@@ -545,7 +553,7 @@ class Blob(_PropertyMixin):
         except NotFound:
             return False
 
-    def delete(self, client=None):
+    def delete(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Deletes a blob from Cloud Storage.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -555,12 +563,20 @@ class Blob(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the blob's bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :raises: :class:`google.cloud.exceptions.NotFound`
                  (propagated from
                  :meth:`google.cloud.storage.bucket.Bucket.delete_blob`).
         """
-        self.bucket.delete_blob(self.name, client=client, generation=self.generation)
+        self.bucket.delete_blob(
+            self.name, client=client, generation=self.generation, timeout=timeout
+        )
 
     def _get_transport(self, client):
         """Return the client's transport.
@@ -1464,7 +1480,9 @@ class Blob(_PropertyMixin):
         except resumable_media.InvalidResponse as exc:
             _raise_from_invalid_response(exc)
 
-    def get_iam_policy(self, client=None, requested_policy_version=None):
+    def get_iam_policy(
+        self, client=None, requested_policy_version=None, timeout=_DEFAULT_TIMEOUT
+    ):
         """Retrieve the IAM policy for the object.
 
         .. note:
@@ -1494,6 +1512,12 @@ class Blob(_PropertyMixin):
                                          The service might return a policy with version lower
                                          than the one that was requested, based on the
                                          feature syntax in the policy fetched.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
@@ -1514,10 +1538,11 @@ class Blob(_PropertyMixin):
             path="%s/iam" % (self.path,),
             query_params=query_params,
             _target_object=None,
+            timeout=timeout,
         )
         return Policy.from_api_repr(info)
 
-    def set_iam_policy(self, policy, client=None):
+    def set_iam_policy(self, policy, client=None, timeout=_DEFAULT_TIMEOUT):
         """Update the IAM policy for the bucket.
 
         .. note:
@@ -1538,6 +1563,12 @@ class Blob(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
@@ -1558,10 +1589,11 @@ class Blob(_PropertyMixin):
             query_params=query_params,
             data=resource,
             _target_object=None,
+            timeout=timeout,
         )
         return Policy.from_api_repr(info)
 
-    def test_iam_permissions(self, permissions, client=None):
+    def test_iam_permissions(self, permissions, client=None, timeout=_DEFAULT_TIMEOUT):
         """API call:  test permissions
 
         .. note:
@@ -1582,6 +1614,12 @@ class Blob(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: list of string
         :returns: the permissions returned by the ``testIamPermissions`` API
@@ -1595,7 +1633,7 @@ class Blob(_PropertyMixin):
 
         path = "%s/iam/testPermissions" % (self.path,)
         resp = client._connection.api_request(
-            method="GET", path=path, query_params=query_params
+            method="GET", path=path, query_params=query_params, timeout=timeout
         )
 
         return resp.get("permissions", [])
@@ -1622,7 +1660,7 @@ class Blob(_PropertyMixin):
         self.acl.all().revoke_read()
         self.acl.save(client=client)
 
-    def compose(self, sources, client=None):
+    def compose(self, sources, client=None, timeout=_DEFAULT_TIMEOUT):
         """Concatenate source blobs into this one.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -1635,6 +1673,12 @@ class Blob(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the blob's bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         client = self._require_client(client)
         query_params = {}
@@ -1652,10 +1696,11 @@ class Blob(_PropertyMixin):
             query_params=query_params,
             data=request,
             _target_object=self,
+            timeout=timeout,
         )
         self._set_properties(api_response)
 
-    def rewrite(self, source, token=None, client=None):
+    def rewrite(self, source, token=None, client=None, timeout=_DEFAULT_TIMEOUT):
         """Rewrite source blob into this one.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -1674,6 +1719,13 @@ class Blob(_PropertyMixin):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the blob's bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: tuple
         :returns: ``(token, bytes_rewritten, total_bytes)``, where ``token``
                   is a rewrite token (``None`` if the rewrite is complete),
@@ -1705,6 +1757,7 @@ class Blob(_PropertyMixin):
             data=self._properties,
             headers=headers,
             _target_object=self,
+            timeout=timeout,
         )
         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 140fc94..4914844 100644
--- a/google/cloud/storage/bucket.py
+++ b/google/cloud/storage/bucket.py
@@ -17,6 +17,7 @@
 import base64
 import copy
 import datetime
+import functools
 import json
 import warnings
 
@@ -39,6 +40,7 @@ from google.cloud.storage._signing import generate_signed_url_v4
 from google.cloud.storage.acl import BucketACL
 from google.cloud.storage.acl import DefaultObjectACL
 from google.cloud.storage.blob import Blob
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 from google.cloud.storage.constants import ARCHIVE_STORAGE_CLASS
 from google.cloud.storage.constants import COLDLINE_STORAGE_CLASS
 from google.cloud.storage.constants import DUAL_REGION_LOCATION_TYPE
@@ -638,7 +640,7 @@ class Bucket(_PropertyMixin):
             payload_format=payload_format,
         )
 
-    def exists(self, client=None):
+    def exists(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Determines whether or not this bucket exists.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -647,6 +649,12 @@ class Bucket(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: bool
         :returns: True if the bucket exists in Cloud Storage.
@@ -667,6 +675,7 @@ class Bucket(_PropertyMixin):
                 path=self.path,
                 query_params=query_params,
                 _target_object=None,
+                timeout=timeout,
             )
             # NOTE: This will not fail immediately in a batch. However, when
             #       Batch.finish() is called, the resulting `NotFound` will be
@@ -682,6 +691,7 @@ class Bucket(_PropertyMixin):
         location=None,
         predefined_acl=None,
         predefined_default_object_acl=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """Creates current bucket.
 
@@ -719,6 +729,13 @@ class Bucket(_PropertyMixin):
         :param predefined_default_object_acl:
             Optional. Name of predefined ACL to apply to bucket's objects. See:
             https://cloud.google.com/storage/docs/access-control/lists#predefined-acl
+
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         if self.user_project is not None:
             raise ValueError("Cannot create bucket with 'user_project' set.")
@@ -755,10 +772,11 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             data=properties,
             _target_object=self,
+            timeout=timeout,
         )
         self._set_properties(api_response)
 
-    def patch(self, client=None):
+    def patch(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Sends all changed properties in a PATCH request.
 
         Updates the ``_properties`` with the response from the backend.
@@ -769,6 +787,12 @@ class Bucket(_PropertyMixin):
                       ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current object.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         # Special case: For buckets, it is possible that labels are being
         # removed; this requires special handling.
@@ -779,7 +803,7 @@ class Bucket(_PropertyMixin):
                 self._properties["labels"][removed_label] = None
 
         # Call the superclass method.
-        return super(Bucket, self).patch(client=client)
+        return super(Bucket, self).patch(client=client, timeout=timeout)
 
     @property
     def acl(self):
@@ -812,7 +836,13 @@ class Bucket(_PropertyMixin):
         return self.path_helper(self.name)
 
     def get_blob(
-        self, blob_name, client=None, encryption_key=None, generation=None, **kwargs
+        self,
+        blob_name,
+        client=None,
+        encryption_key=None,
+        generation=None,
+        timeout=_DEFAULT_TIMEOUT,
+        **kwargs
     ):
         """Get a blob object by name.
 
@@ -842,6 +872,13 @@ class Bucket(_PropertyMixin):
         :param generation: Optional. If present, selects a specific revision of
                            this object.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :param kwargs: Keyword arguments to pass to the
                        :class:`~google.cloud.storage.blob.Blob` constructor.
 
@@ -859,7 +896,7 @@ class Bucket(_PropertyMixin):
             # NOTE: This will not fail immediately in a batch. However, when
             #       Batch.finish() is called, the resulting `NotFound` will be
             #       raised.
-            blob.reload(client=client)
+            blob.reload(client=client, timeout=timeout)
         except NotFound:
             return None
         else:
@@ -875,6 +912,7 @@ class Bucket(_PropertyMixin):
         projection="noAcl",
         fields=None,
         client=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -924,6 +962,13 @@ class Bucket(_PropertyMixin):
         :param client: (Optional) The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :returns: Iterator of all :class:`~google.cloud.storage.blob.Blob`
                   in this bucket matching the arguments.
@@ -947,9 +992,10 @@ class Bucket(_PropertyMixin):
 
         client = self._require_client(client)
         path = self.path + "/o"
+        api_request = functools.partial(client._connection.api_request, timeout=timeout)
         iterator = page_iterator.HTTPIterator(
             client=client,
-            api_request=client._connection.api_request,
+            api_request=api_request,
             path=path,
             item_to_value=_item_to_blob,
             page_token=page_token,
@@ -961,7 +1007,7 @@ class Bucket(_PropertyMixin):
         iterator.prefixes = set()
         return iterator
 
-    def list_notifications(self, client=None):
+    def list_notifications(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """List Pub / Sub notifications for this bucket.
 
         See:
@@ -973,22 +1019,29 @@ class Bucket(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: list of :class:`.BucketNotification`
         :returns: notification instances
         """
         client = self._require_client(client)
         path = self.path + "/notificationConfigs"
+        api_request = functools.partial(client._connection.api_request, timeout=timeout)
         iterator = page_iterator.HTTPIterator(
             client=client,
-            api_request=client._connection.api_request,
+            api_request=api_request,
             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, timeout=_DEFAULT_TIMEOUT):
         """Delete this bucket.
 
         The bucket **must** be empty in order to submit a delete request. If
@@ -1014,6 +1067,12 @@ class Bucket(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response on each request.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :raises: :class:`ValueError` if ``force`` is ``True`` and the bucket
                  contains more than 256 objects / blobs.
@@ -1027,7 +1086,9 @@ class Bucket(_PropertyMixin):
         if force:
             blobs = list(
                 self.list_blobs(
-                    max_results=self._MAX_OBJECTS_FOR_ITERATION + 1, client=client
+                    max_results=self._MAX_OBJECTS_FOR_ITERATION + 1,
+                    client=client,
+                    timeout=timeout,
                 )
             )
             if len(blobs) > self._MAX_OBJECTS_FOR_ITERATION:
@@ -1040,7 +1101,9 @@ class Bucket(_PropertyMixin):
                 raise ValueError(message)
 
             # Ignore 404 errors on delete.
-            self.delete_blobs(blobs, on_error=lambda blob: None, client=client)
+            self.delete_blobs(
+                blobs, on_error=lambda blob: None, client=client, timeout=timeout
+            )
 
         # We intentionally pass `_target_object=None` since a DELETE
         # request has no response value (whether in a standard request or
@@ -1050,9 +1113,12 @@ class Bucket(_PropertyMixin):
             path=self.path,
             query_params=query_params,
             _target_object=None,
+            timeout=timeout,
         )
 
-    def delete_blob(self, blob_name, client=None, generation=None):
+    def delete_blob(
+        self, blob_name, client=None, generation=None, timeout=_DEFAULT_TIMEOUT
+    ):
         """Deletes a blob from the current bucket.
 
         If the blob isn't found (backend 404), raises a
@@ -1078,6 +1144,13 @@ class Bucket(_PropertyMixin):
         :param generation: Optional. If present, permanently deletes a specific
                            revision of this object.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises: :class:`google.cloud.exceptions.NotFound` (to suppress
                  the exception, call ``delete_blobs``, passing a no-op
                  ``on_error`` callback, e.g.:
@@ -1098,9 +1171,10 @@ class Bucket(_PropertyMixin):
             path=blob.path,
             query_params=blob._query_params,
             _target_object=None,
+            timeout=timeout,
         )
 
-    def delete_blobs(self, blobs, on_error=None, client=None):
+    def delete_blobs(self, blobs, on_error=None, client=None, timeout=_DEFAULT_TIMEOUT):
         """Deletes a list of blobs from the current bucket.
 
         Uses :meth:`delete_blob` to delete each individual blob.
@@ -1121,6 +1195,14 @@ class Bucket(_PropertyMixin):
         :param client: (Optional) The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response. The timeout applies to each individual
+            blob delete request.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises: :class:`~google.cloud.exceptions.NotFound` (if
                  `on_error` is not passed).
         """
@@ -1129,7 +1211,7 @@ class Bucket(_PropertyMixin):
                 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, timeout=timeout)
             except NotFound:
                 if on_error is not None:
                     on_error(blob)
@@ -1144,6 +1226,7 @@ class Bucket(_PropertyMixin):
         client=None,
         preserve_acl=True,
         source_generation=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """Copy the given blob to the given bucket, optionally with a new name.
 
@@ -1172,6 +1255,13 @@ class Bucket(_PropertyMixin):
         :param source_generation: Optional. The generation of the blob to be
                                   copied.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: :class:`google.cloud.storage.blob.Blob`
         :returns: The new Blob.
         """
@@ -1194,15 +1284,16 @@ class Bucket(_PropertyMixin):
             path=api_path,
             query_params=query_params,
             _target_object=new_blob,
+            timeout=timeout,
         )
 
         if not preserve_acl:
-            new_blob.acl.save(acl={}, client=client)
+            new_blob.acl.save(acl={}, client=client, timeout=timeout)
 
         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, timeout=_DEFAULT_TIMEOUT):
         """Rename the given blob using copy and delete operations.
 
         If :attr:`user_project` is set, bills the API request to that project.
@@ -1227,15 +1318,23 @@ class Bucket(_PropertyMixin):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response. The timeout applies to each individual
+            request.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :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, timeout=timeout)
 
         if not same_name:
-            blob.delete(client=client)
+            blob.delete(client=client, timeout=timeout)
 
         return new_blob
 
@@ -1863,7 +1962,9 @@ class Bucket(_PropertyMixin):
         """
         return self.configure_website(None, None)
 
-    def get_iam_policy(self, client=None, requested_policy_version=None):
+    def get_iam_policy(
+        self, client=None, requested_policy_version=None, timeout=_DEFAULT_TIMEOUT
+    ):
         """Retrieve the IAM policy for the bucket.
 
         See
@@ -1888,6 +1989,13 @@ class Bucket(_PropertyMixin):
                                          than the one that was requested, based on the
                                          feature syntax in the policy fetched.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
                   the ``getIamPolicy`` API request.
@@ -1930,10 +2038,11 @@ class Bucket(_PropertyMixin):
             path="%s/iam" % (self.path,),
             query_params=query_params,
             _target_object=None,
+            timeout=timeout,
         )
         return Policy.from_api_repr(info)
 
-    def set_iam_policy(self, policy, client=None):
+    def set_iam_policy(self, policy, client=None, timeout=_DEFAULT_TIMEOUT):
         """Update the IAM policy for the bucket.
 
         See
@@ -1949,6 +2058,13 @@ class Bucket(_PropertyMixin):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: :class:`google.api_core.iam.Policy`
         :returns: the policy instance, based on the resource returned from
                   the ``setIamPolicy`` API request.
@@ -1967,10 +2083,11 @@ class Bucket(_PropertyMixin):
             query_params=query_params,
             data=resource,
             _target_object=None,
+            timeout=timeout,
         )
         return Policy.from_api_repr(info)
 
-    def test_iam_permissions(self, permissions, client=None):
+    def test_iam_permissions(self, permissions, client=None, timeout=_DEFAULT_TIMEOUT):
         """API call:  test permissions
 
         See
@@ -1986,6 +2103,13 @@ class Bucket(_PropertyMixin):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: list of string
         :returns: the permissions returned by the ``testIamPermissions`` API
                   request.
@@ -1998,11 +2122,13 @@ class Bucket(_PropertyMixin):
 
         path = "%s/iam/testPermissions" % (self.path,)
         resp = client._connection.api_request(
-            method="GET", path=path, query_params=query_params
+            method="GET", path=path, query_params=query_params, timeout=timeout
         )
         return resp.get("permissions", [])
 
-    def make_public(self, recursive=False, future=False, client=None):
+    def make_public(
+        self, recursive=False, future=False, client=None, timeout=_DEFAULT_TIMEOUT
+    ):
         """Update bucket's ACL, granting read access to anonymous users.
 
         :type recursive: bool
@@ -2017,6 +2143,13 @@ class Bucket(_PropertyMixin):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response. The timeout applies to each underlying
+            request.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :raises ValueError:
             If ``recursive`` is True, and the bucket contains more than 256
@@ -2027,14 +2160,14 @@ class Bucket(_PropertyMixin):
             for each blob.
         """
         self.acl.all().grant_read()
-        self.acl.save(client=client)
+        self.acl.save(client=client, timeout=timeout)
 
         if future:
             doa = self.default_object_acl
             if not doa.loaded:
-                doa.reload(client=client)
+                doa.reload(client=client, timeout=timeout)
             doa.all().grant_read()
-            doa.save(client=client)
+            doa.save(client=client, timeout=timeout)
 
         if recursive:
             blobs = list(
@@ -2042,6 +2175,7 @@ class Bucket(_PropertyMixin):
                     projection="full",
                     max_results=self._MAX_OBJECTS_FOR_ITERATION + 1,
                     client=client,
+                    timeout=timeout,
                 )
             )
             if len(blobs) > self._MAX_OBJECTS_FOR_ITERATION:
@@ -2056,9 +2190,11 @@ class Bucket(_PropertyMixin):
 
             for blob in blobs:
                 blob.acl.all().grant_read()
-                blob.acl.save(client=client)
+                blob.acl.save(client=client, timeout=timeout)
 
-    def make_private(self, recursive=False, future=False, client=None):
+    def make_private(
+        self, recursive=False, future=False, client=None, timeout=_DEFAULT_TIMEOUT
+    ):
         """Update bucket's ACL, revoking read access for anonymous users.
 
         :type recursive: bool
@@ -2074,6 +2210,14 @@ class Bucket(_PropertyMixin):
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response. The timeout applies to each underlying
+            request.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises ValueError:
             If ``recursive`` is True, and the bucket contains more than 256
             blobs.  This is to prevent extremely long runtime of this
@@ -2083,14 +2227,14 @@ class Bucket(_PropertyMixin):
             for each blob.
         """
         self.acl.all().revoke_read()
-        self.acl.save(client=client)
+        self.acl.save(client=client, timeout=timeout)
 
         if future:
             doa = self.default_object_acl
             if not doa.loaded:
-                doa.reload(client=client)
+                doa.reload(client=client, timeout=timeout)
             doa.all().revoke_read()
-            doa.save(client=client)
+            doa.save(client=client, timeout=timeout)
 
         if recursive:
             blobs = list(
@@ -2098,6 +2242,7 @@ class Bucket(_PropertyMixin):
                     projection="full",
                     max_results=self._MAX_OBJECTS_FOR_ITERATION + 1,
                     client=client,
+                    timeout=timeout,
                 )
             )
             if len(blobs) > self._MAX_OBJECTS_FOR_ITERATION:
@@ -2112,7 +2257,7 @@ class Bucket(_PropertyMixin):
 
             for blob in blobs:
                 blob.acl.all().revoke_read()
-                blob.acl.save(client=client)
+                blob.acl.save(client=client, timeout=timeout)
 
     def generate_upload_policy(self, conditions, expiration=None, client=None):
         """Create a signed upload policy for uploading objects.
@@ -2176,9 +2321,16 @@ class Bucket(_PropertyMixin):
 
         return fields
 
-    def lock_retention_policy(self, client=None):
+    def lock_retention_policy(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Lock the bucket's retention policy.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises ValueError:
             if the bucket has no metageneration (i.e., new or never reloaded);
             if the bucket has no retention policy assigned;
@@ -2204,7 +2356,11 @@ class Bucket(_PropertyMixin):
 
         path = "/b/{}/lockRetentionPolicy".format(self.name)
         api_response = client._connection.api_request(
-            method="POST", path=path, query_params=query_params, _target_object=self
+            method="POST",
+            path=path,
+            query_params=query_params,
+            _target_object=self,
+            timeout=timeout,
         )
         self._set_properties(api_response)
 
diff --git a/google/cloud/storage/client.py b/google/cloud/storage/client.py
index 1d51e65..41c1238 100644
--- a/google/cloud/storage/client.py
+++ b/google/cloud/storage/client.py
@@ -14,6 +14,8 @@
 
 """Client for interacting with the Google Cloud Storage API."""
 
+import functools
+
 import google.api_core.client_options
 
 from google.auth.credentials import AnonymousCredentials
@@ -30,6 +32,7 @@ from google.cloud.storage.blob import Blob
 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
 
 
 _marker = object()
@@ -211,13 +214,19 @@ class Client(ClientWithProject):
         """
         return self._batch_stack.top
 
-    def get_service_account_email(self, project=None):
+    def get_service_account_email(self, project=None, timeout=_DEFAULT_TIMEOUT):
         """Get the email address of the project's GCS service account
 
         :type project: str
         :param project:
             (Optional) Project ID to use for retreiving GCS service account
             email address.  Defaults to the client's project.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: str
         :returns: service account email address
@@ -225,7 +234,9 @@ class Client(ClientWithProject):
         if project is None:
             project = self.project
         path = "/projects/%s/serviceAccount" % (project,)
-        api_response = self._base_connection.api_request(method="GET", path=path)
+        api_response = self._base_connection.api_request(
+            method="GET", path=path, timeout=timeout
+        )
         return api_response["email_address"]
 
     def bucket(self, bucket_name, user_project=None):
@@ -259,7 +270,7 @@ class Client(ClientWithProject):
         """
         return Batch(client=self)
 
-    def get_bucket(self, bucket_or_name):
+    def get_bucket(self, bucket_or_name, timeout=_DEFAULT_TIMEOUT):
         """API call: retrieve a bucket via a GET request.
 
         See
@@ -272,6 +283,12 @@ class Client(ClientWithProject):
             ]):
                 The bucket resource to pass or name to create.
 
+            timeout (Optional[Union[float, Tuple[float, float]]]):
+                The amount of time, in seconds, to wait for the server response.
+
+                Can also be passed as a tuple (connect_timeout, read_timeout).
+                See :meth:`requests.Session.request` documentation for details.
+
         Returns:
             google.cloud.storage.bucket.Bucket
                 The bucket matching the name provided.
@@ -302,10 +319,10 @@ class Client(ClientWithProject):
         """
         bucket = self._bucket_arg_to_bucket(bucket_or_name)
 
-        bucket.reload(client=self)
+        bucket.reload(client=self, timeout=timeout)
         return bucket
 
-    def lookup_bucket(self, bucket_name):
+    def lookup_bucket(self, bucket_name, timeout=_DEFAULT_TIMEOUT):
         """Get a bucket by name, returning None if not found.
 
         You can use this if you would rather check for a None value
@@ -318,11 +335,18 @@ class Client(ClientWithProject):
         :type bucket_name: str
         :param bucket_name: The name of the bucket to get.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: :class:`google.cloud.storage.bucket.Bucket`
         :returns: The bucket matching the name provided or None if not found.
         """
         try:
-            return self.get_bucket(bucket_name)
+            return self.get_bucket(bucket_name, timeout=timeout)
         except NotFound:
             return None
 
@@ -335,6 +359,7 @@ class Client(ClientWithProject):
         location=None,
         predefined_acl=None,
         predefined_default_object_acl=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """API call: create a new bucket via a POST request.
 
@@ -366,6 +391,11 @@ class Client(ClientWithProject):
             predefined_default_object_acl (str):
                 Optional. Name of predefined ACL to apply to bucket's objects. See:
                 https://cloud.google.com/storage/docs/access-control/lists#predefined-acl
+            timeout (Optional[Union[float, Tuple[float, float]]]):
+                The amount of time, in seconds, to wait for the server response.
+
+                Can also be passed as a tuple (connect_timeout, read_timeout).
+                See :meth:`requests.Session.request` documentation for details.
 
         Returns:
             google.cloud.storage.bucket.Bucket
@@ -434,6 +464,7 @@ class Client(ClientWithProject):
             query_params=query_params,
             data=properties,
             _target_object=bucket,
+            timeout=timeout,
         )
 
         bucket._set_properties(api_response)
@@ -495,6 +526,7 @@ class Client(ClientWithProject):
         versions=None,
         projection="noAcl",
         fields=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """Return an iterator used to find blobs in the bucket.
 
@@ -541,6 +573,12 @@ class Client(ClientWithProject):
                 ``'items(name,contentLanguage),nextPageToken'``.
                 See: https://cloud.google.com/storage/docs/json_api/v1/parameters#fields
 
+            timeout (Optional[Union[float, Tuple[float, float]]]):
+                The amount of time, in seconds, to wait for the server response.
+
+                Can also be passed as a tuple (connect_timeout, read_timeout).
+                See :meth:`requests.Session.request` documentation for details.
+
         Returns:
             Iterator of all :class:`~google.cloud.storage.blob.Blob`
             in this bucket matching the arguments.
@@ -555,6 +593,7 @@ class Client(ClientWithProject):
             projection=projection,
             fields=fields,
             client=self,
+            timeout=timeout,
         )
 
     def list_buckets(
@@ -565,6 +604,7 @@ class Client(ClientWithProject):
         projection="noAcl",
         fields=None,
         project=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """Get all buckets in the project associated to the client.
 
@@ -608,6 +648,13 @@ class Client(ClientWithProject):
         :param project: (Optional) the project whose buckets are to be listed.
                         If not passed, uses the project set on the client.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: :class:`~google.api_core.page_iterator.Iterator`
         :raises ValueError: if both ``project`` is ``None`` and the client's
                             project is also ``None``.
@@ -630,9 +677,11 @@ class Client(ClientWithProject):
         if fields is not None:
             extra_params["fields"] = fields
 
+        api_request = functools.partial(self._connection.api_request, timeout=timeout)
+
         return page_iterator.HTTPIterator(
             client=self,
-            api_request=self._connection.api_request,
+            api_request=api_request,
             path="/b",
             item_to_value=_item_to_bucket,
             page_token=page_token,
@@ -641,7 +690,11 @@ class Client(ClientWithProject):
         )
 
     def create_hmac_key(
-        self, service_account_email, project_id=None, user_project=None
+        self,
+        service_account_email,
+        project_id=None,
+        user_project=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """Create an HMAC key for a service account.
 
@@ -655,6 +708,13 @@ class Client(ClientWithProject):
         :type user_project: str
         :param user_project: (Optional) This parameter is currently ignored.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype:
             Tuple[:class:`~google.cloud.storage.hmac_key.HMACKeyMetadata`, str]
         :returns: metadata for the created key, plus the bytes of the key's secret, which is an 40-character base64-encoded string.
@@ -669,7 +729,7 @@ class Client(ClientWithProject):
             qs_params["userProject"] = user_project
 
         api_response = self._connection.api_request(
-            method="POST", path=path, query_params=qs_params
+            method="POST", path=path, query_params=qs_params, timeout=timeout
         )
         metadata = HMACKeyMetadata(self)
         metadata._properties = api_response["metadata"]
@@ -683,6 +743,7 @@ class Client(ClientWithProject):
         show_deleted_keys=None,
         project_id=None,
         user_project=None,
+        timeout=_DEFAULT_TIMEOUT,
     ):
         """List HMAC keys for a project.
 
@@ -706,6 +767,13 @@ class Client(ClientWithProject):
         :type user_project: str
         :param user_project: (Optional) This parameter is currently ignored.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype:
             Tuple[:class:`~google.cloud.storage.hmac_key.HMACKeyMetadata`, str]
         :returns: metadata for the created key, plus the bytes of the key's secret, which is an 40-character base64-encoded string.
@@ -725,16 +793,20 @@ class Client(ClientWithProject):
         if user_project is not None:
             extra_params["userProject"] = user_project
 
+        api_request = functools.partial(self._connection.api_request, timeout=timeout)
+
         return page_iterator.HTTPIterator(
             client=self,
-            api_request=self._connection.api_request,
+            api_request=api_request,
             path=path,
             item_to_value=_item_to_hmac_key_metadata,
             max_results=max_results,
             extra_params=extra_params,
         )
 
-    def get_hmac_key_metadata(self, access_id, project_id=None, user_project=None):
+    def get_hmac_key_metadata(
+        self, access_id, project_id=None, user_project=None, timeout=_DEFAULT_TIMEOUT
+    ):
         """Return a metadata instance for the given HMAC key.
 
         :type access_id: str
@@ -744,11 +816,18 @@ class Client(ClientWithProject):
         :param project_id: (Optional) project ID of an existing key.
             Defaults to client's project.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :type user_project: str
         :param user_project: (Optional) This parameter is currently ignored.
         """
         metadata = HMACKeyMetadata(self, access_id, project_id, user_project)
-        metadata.reload()  # raises NotFound for missing key
+        metadata.reload(timeout=timeout)  # raises NotFound for missing key
         return metadata
 
 
diff --git a/google/cloud/storage/constants.py b/google/cloud/storage/constants.py
index faadff1..6215086 100644
--- a/google/cloud/storage/constants.py
+++ b/google/cloud/storage/constants.py
@@ -89,3 +89,10 @@ DUAL_REGION_LOCATION_TYPE = "dual-region"
 
 Provides high availability and low latency across two regions.
 """
+
+
+# Internal constants
+
+_DEFAULT_TIMEOUT = 60  # in seconds
+"""The default request timeout in seconds if a timeout is not explicitly given.
+"""
diff --git a/google/cloud/storage/hmac_key.py b/google/cloud/storage/hmac_key.py
index 96ccbca..296b38e 100644
--- a/google/cloud/storage/hmac_key.py
+++ b/google/cloud/storage/hmac_key.py
@@ -15,6 +15,8 @@
 from google.cloud.exceptions import NotFound
 from google.cloud._helpers import _rfc3339_to_datetime
 
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
 
 class HMACKeyMetadata(object):
     """Metadata about an HMAC service account key withn Cloud Storage.
@@ -185,9 +187,16 @@ class HMACKeyMetadata(object):
         """
         return self._user_project
 
-    def exists(self):
+    def exists(self, timeout=_DEFAULT_TIMEOUT):
         """Determine whether or not the key for this metadata exists.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :rtype: bool
         :returns: True if the key exists in Cloud Storage.
         """
@@ -198,16 +207,23 @@ class HMACKeyMetadata(object):
                 qs_params["userProject"] = self.user_project
 
             self._client._connection.api_request(
-                method="GET", path=self.path, query_params=qs_params
+                method="GET", path=self.path, query_params=qs_params, timeout=timeout
             )
         except NotFound:
             return False
         else:
             return True
 
-    def reload(self):
+    def reload(self, timeout=_DEFAULT_TIMEOUT):
         """Reload properties from Cloud Storage.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises :class:`~google.api_core.exceptions.NotFound`:
             if the key does not exist on the back-end.
         """
@@ -217,12 +233,19 @@ class HMACKeyMetadata(object):
             qs_params["userProject"] = self.user_project
 
         self._properties = self._client._connection.api_request(
-            method="GET", path=self.path, query_params=qs_params
+            method="GET", path=self.path, query_params=qs_params, timeout=timeout
         )
 
-    def update(self):
+    def update(self, timeout=_DEFAULT_TIMEOUT):
         """Save writable properties to Cloud Storage.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises :class:`~google.api_core.exceptions.NotFound`:
             if the key does not exist on the back-end.
         """
@@ -232,12 +255,23 @@ class HMACKeyMetadata(object):
 
         payload = {"state": self.state}
         self._properties = self._client._connection.api_request(
-            method="PUT", path=self.path, data=payload, query_params=qs_params
+            method="PUT",
+            path=self.path,
+            data=payload,
+            query_params=qs_params,
+            timeout=timeout,
         )
 
-    def delete(self):
+    def delete(self, timeout=_DEFAULT_TIMEOUT):
         """Delete the key from Cloud Storage.
 
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
+
         :raises :class:`~google.api_core.exceptions.NotFound`:
             if the key does not exist on the back-end.
         """
@@ -249,5 +283,5 @@ class HMACKeyMetadata(object):
             qs_params["userProject"] = self.user_project
 
         self._client._connection.api_request(
-            method="DELETE", path=self.path, query_params=qs_params
+            method="DELETE", path=self.path, query_params=qs_params, timeout=timeout
         )
diff --git a/google/cloud/storage/notification.py b/google/cloud/storage/notification.py
index 982dc16..e9618f6 100644
--- a/google/cloud/storage/notification.py
+++ b/google/cloud/storage/notification.py
@@ -18,6 +18,8 @@ import re
 
 from google.api_core.exceptions import NotFound
 
+from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
 
 OBJECT_FINALIZE_EVENT_TYPE = "OBJECT_FINALIZE"
 OBJECT_METADATA_UPDATE_EVENT_TYPE = "OBJECT_METADATA_UPDATE"
@@ -221,7 +223,7 @@ class BucketNotification(object):
         self._properties.clear()
         self._properties.update(response)
 
-    def create(self, client=None):
+    def create(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """API wrapper: create the notification.
 
         See:
@@ -233,6 +235,12 @@ class BucketNotification(object):
         :type client: :class:`~google.cloud.storage.client.Client`
         :param client: (Optional) the client to use.  If not passed, falls back
                        to the ``client`` stored on the notification's bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
         """
         if self.notification_id is not None:
             raise ValueError(
@@ -249,10 +257,14 @@ class BucketNotification(object):
         properties = self._properties.copy()
         properties["topic"] = _TOPIC_REF_FMT.format(self.topic_project, self.topic_name)
         self._properties = client._connection.api_request(
-            method="POST", path=path, query_params=query_params, data=properties
+            method="POST",
+            path=path,
+            query_params=query_params,
+            data=properties,
+            timeout=timeout,
         )
 
-    def exists(self, client=None):
+    def exists(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Test whether this notification exists.
 
         See:
@@ -265,6 +277,12 @@ class BucketNotification(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: bool
         :returns: True, if the notification exists, else False.
@@ -281,14 +299,14 @@ class BucketNotification(object):
 
         try:
             client._connection.api_request(
-                method="GET", path=self.path, query_params=query_params
+                method="GET", path=self.path, query_params=query_params, timeout=timeout
             )
         except NotFound:
             return False
         else:
             return True
 
-    def reload(self, client=None):
+    def reload(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Update this notification from the server configuration.
 
         See:
@@ -301,6 +319,12 @@ class BucketNotification(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :rtype: bool
         :returns: True, if the notification exists, else False.
@@ -316,11 +340,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
+            method="GET", path=self.path, query_params=query_params, timeout=timeout
         )
         self._set_properties(response)
 
-    def delete(self, client=None):
+    def delete(self, client=None, timeout=_DEFAULT_TIMEOUT):
         """Delete this notification.
 
         See:
@@ -333,6 +357,12 @@ class BucketNotification(object):
                       ``NoneType``
         :param client: Optional. The client to use.  If not passed, falls back
                        to the ``client`` stored on the current bucket.
+        :type timeout: float or tuple
+        :param timeout: (optional) The amount of time, in seconds, to wait
+            for the server response.
+
+            Can also be passed as a tuple (connect_timeout, read_timeout).
+            See :meth:`requests.Session.request` documentation for details.
 
         :raises: :class:`google.api_core.exceptions.NotFound`:
             if the notification does not exist.
@@ -348,7 +378,7 @@ class BucketNotification(object):
             query_params["userProject"] = self.bucket.user_project
 
         client._connection.api_request(
-            method="DELETE", path=self.path, query_params=query_params
+            method="DELETE", path=self.path, query_params=query_params, timeout=timeout
         )
 
 


Test patch:
diff --git a/tests/unit/test__helpers.py b/tests/unit/test__helpers.py
index 9b75b0e..10b71b7 100644
--- a/tests/unit/test__helpers.py
+++ b/tests/unit/test__helpers.py
@@ -44,6 +44,12 @@ class Test__get_storage_host(unittest.TestCase):
 
 
 class Test_PropertyMixin(unittest.TestCase):
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     @staticmethod
     def _get_target_class():
         from google.cloud.storage._helpers import _PropertyMixin
@@ -103,7 +109,7 @@ class Test_PropertyMixin(unittest.TestCase):
         # Make sure changes is not a set instance before calling reload
         # (which will clear / replace it with an empty set), checked below.
         derived._changes = object()
-        derived.reload(client=client)
+        derived.reload(client=client, timeout=42)
         self.assertEqual(derived._properties, {"foo": "Foo"})
         kw = connection._requested
         self.assertEqual(len(kw), 1)
@@ -115,6 +121,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "query_params": {"projection": "noAcl"},
                 "headers": {},
                 "_target_object": derived,
+                "timeout": 42,
             },
         )
         self.assertEqual(derived._changes, set())
@@ -139,6 +146,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 "query_params": {"projection": "noAcl", "userProject": user_project},
                 "headers": {},
                 "_target_object": derived,
+                "timeout": self._get_default_timeout(),
             },
         )
         self.assertEqual(derived._changes, set())
@@ -164,7 +172,7 @@ class Test_PropertyMixin(unittest.TestCase):
         BAZ = object()
         derived._properties = {"bar": BAR, "baz": BAZ}
         derived._changes = set(["bar"])  # Ignore baz.
-        derived.patch(client=client)
+        derived.patch(client=client, timeout=42)
         self.assertEqual(derived._properties, {"foo": "Foo"})
         kw = connection._requested
         self.assertEqual(len(kw), 1)
@@ -177,6 +185,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 # Since changes does not include `baz`, we don't see it sent.
                 "data": {"bar": BAR},
                 "_target_object": derived,
+                "timeout": 42,
             },
         )
         # Make sure changes get reset by patch().
@@ -205,6 +214,7 @@ class Test_PropertyMixin(unittest.TestCase):
                 # Since changes does not include `baz`, we don't see it sent.
                 "data": {"bar": BAR},
                 "_target_object": derived,
+                "timeout": self._get_default_timeout(),
             },
         )
         # Make sure changes get reset by patch().
@@ -219,7 +229,7 @@ class Test_PropertyMixin(unittest.TestCase):
         BAZ = object()
         derived._properties = {"bar": BAR, "baz": BAZ}
         derived._changes = set(["bar"])  # Update sends 'baz' anyway.
-        derived.update(client=client)
+        derived.update(client=client, timeout=42)
         self.assertEqual(derived._properties, {"foo": "Foo"})
         kw = connection._requested
         self.assertEqual(len(kw), 1)
@@ -227,6 +237,7 @@ class Test_PropertyMixin(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/path")
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
         self.assertEqual(kw[0]["data"], {"bar": BAR, "baz": BAZ})
+        self.assertEqual(kw[0]["timeout"], 42)
         # Make sure changes get reset by patch().
         self.assertEqual(derived._changes, set())
 
@@ -250,6 +261,7 @@ class Test_PropertyMixin(unittest.TestCase):
             kw[0]["query_params"], {"projection": "full", "userProject": user_project}
         )
         self.assertEqual(kw[0]["data"], {"bar": BAR, "baz": BAZ})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
         # 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 2100918..021698e 100644
--- a/tests/unit/test__http.py
+++ b/tests/unit/test__http.py
@@ -30,6 +30,7 @@ class TestConnection(unittest.TestCase):
     def test_extra_headers(self):
         import requests
         from google.cloud import _http as base_http
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
 
         http = mock.create_autospec(requests.Session, instance=True)
         response = requests.Response()
@@ -55,7 +56,7 @@ class TestConnection(unittest.TestCase):
             headers=expected_headers,
             method="GET",
             url=expected_uri,
-            timeout=base_http._DEFAULT_TIMEOUT,
+            timeout=_DEFAULT_TIMEOUT,
         )
 
     def test_build_api_url_no_extra_query_params(self):
diff --git a/tests/unit/test_acl.py b/tests/unit/test_acl.py
index d66a943..47400f1 100644
--- a/tests/unit/test_acl.py
+++ b/tests/unit/test_acl.py
@@ -125,6 +125,18 @@ class Test_ACLEntity(unittest.TestCase):
         self.assertEqual(entity.get_roles(), set())
 
 
+class FakeReload(object):
+    """A callable used for faking the reload() method of an ACL instance."""
+
+    def __init__(self, acl):
+        self.acl = acl
+        self.timeouts_used = []
+
+    def __call__(self, timeout=None):
+        self.acl.loaded = True
+        self.timeouts_used.append(timeout)
+
+
 class Test_ACL(unittest.TestCase):
     @staticmethod
     def _get_target_class():
@@ -132,6 +144,12 @@ class Test_ACL(unittest.TestCase):
 
         return ACL
 
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     def _make_one(self, *args, **kw):
         return self._get_target_class()(*args, **kw)
 
@@ -150,13 +168,14 @@ class Test_ACL(unittest.TestCase):
 
     def test__ensure_loaded(self):
         acl = self._make_one()
+        _reload = FakeReload(acl)
+        acl.reload = _reload
+        acl.loaded = False
 
-        def _reload():
-            acl._really_loaded = True
+        acl._ensure_loaded(timeout=42)
 
-        acl.reload = _reload
-        acl._ensure_loaded()
-        self.assertTrue(acl._really_loaded)
+        self.assertTrue(acl.loaded)
+        self.assertEqual(_reload.timeouts_used[0], 42)
 
     def test_client_is_abstract(self):
         acl = self._make_one()
@@ -179,13 +198,13 @@ class Test_ACL(unittest.TestCase):
 
     def test___iter___empty_lazy(self):
         acl = self._make_one()
-
-        def _reload():
-            acl.loaded = True
+        _reload = FakeReload(acl)
+        acl.loaded = False
 
         acl.reload = _reload
         self.assertEqual(list(acl), [])
         self.assertTrue(acl.loaded)
+        self.assertEqual(_reload.timeouts_used[0], self._get_default_timeout())
 
     def test___iter___non_empty_no_roles(self):
         TYPE = "type"
@@ -263,13 +282,13 @@ class Test_ACL(unittest.TestCase):
 
     def test_has_entity_miss_str_lazy(self):
         acl = self._make_one()
-
-        def _reload():
-            acl.loaded = True
-
+        _reload = FakeReload(acl)
         acl.reload = _reload
+        acl.loaded = False
+
         self.assertFalse(acl.has_entity("nonesuch"))
         self.assertTrue(acl.loaded)
+        self.assertEqual(_reload.timeouts_used[0], self._get_default_timeout())
 
     def test_has_entity_miss_entity(self):
         from google.cloud.storage.acl import _ACLEntity
@@ -304,13 +323,13 @@ class Test_ACL(unittest.TestCase):
 
     def test_get_entity_miss_str_no_default_lazy(self):
         acl = self._make_one()
-
-        def _reload():
-            acl.loaded = True
-
+        _reload = FakeReload(acl)
         acl.reload = _reload
+        acl.loaded = False
+
         self.assertIsNone(acl.get_entity("nonesuch"))
         self.assertTrue(acl.loaded)
+        self.assertEqual(_reload.timeouts_used[0], self._get_default_timeout())
 
     def test_get_entity_miss_entity_no_default(self):
         from google.cloud.storage.acl import _ACLEntity
@@ -380,15 +399,16 @@ class Test_ACL(unittest.TestCase):
         entity.grant(ROLE)
         acl = self._make_one()
 
-        def _reload():
-            acl.loaded = True
-
+        _reload = FakeReload(acl)
         acl.reload = _reload
+        acl.loaded = False
+
         acl.add_entity(entity)
         self.assertTrue(acl.loaded)
         self.assertEqual(list(acl), [{"entity": "type-id", "role": ROLE}])
         self.assertEqual(list(acl.get_entities()), [entity])
         self.assertTrue(acl.loaded)
+        self.assertEqual(_reload.timeouts_used[0], self._get_default_timeout())
 
     def test_add_entity_hit(self):
         from google.cloud.storage.acl import _ACLEntity
@@ -494,13 +514,13 @@ class Test_ACL(unittest.TestCase):
 
     def test_get_entities_empty_lazy(self):
         acl = self._make_one()
-
-        def _reload():
-            acl.loaded = True
-
+        _reload = FakeReload(acl)
         acl.reload = _reload
+        acl.loaded = False
+
         self.assertEqual(acl.get_entities(), [])
         self.assertTrue(acl.loaded)
+        self.assertEqual(_reload.timeouts_used[0], self._get_default_timeout())
 
     def test_get_entities_nonempty(self):
         TYPE = "type"
@@ -519,12 +539,18 @@ class Test_ACL(unittest.TestCase):
         acl.reload_path = "/testing/acl"
         acl.loaded = True
         acl.entity("allUsers", ROLE)
-        acl.reload(client=client)
+        acl.reload(client=client, timeout=42)
         self.assertEqual(list(acl), [])
         kw = connection._requested
         self.assertEqual(len(kw), 1)
         self.assertEqual(
-            kw[0], {"method": "GET", "path": "/testing/acl", "query_params": {}}
+            kw[0],
+            {
+                "method": "GET",
+                "path": "/testing/acl",
+                "query_params": {},
+                "timeout": 42,
+            },
         )
 
     def test_reload_empty_result_clears_local(self):
@@ -543,7 +569,13 @@ class Test_ACL(unittest.TestCase):
         kw = connection._requested
         self.assertEqual(len(kw), 1)
         self.assertEqual(
-            kw[0], {"method": "GET", "path": "/testing/acl", "query_params": {}}
+            kw[0],
+            {
+                "method": "GET",
+                "path": "/testing/acl",
+                "query_params": {},
+                "timeout": self._get_default_timeout(),
+            },
         )
 
     def test_reload_nonempty_result_w_user_project(self):
@@ -568,6 +600,7 @@ class Test_ACL(unittest.TestCase):
                 "method": "GET",
                 "path": "/testing/acl",
                 "query_params": {"userProject": USER_PROJECT},
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -586,7 +619,7 @@ class Test_ACL(unittest.TestCase):
         acl = self._make_one()
         acl.save_path = "/testing"
         acl.loaded = True
-        acl.save(client=client)
+        acl.save(client=client, timeout=42)
         self.assertEqual(list(acl), [])
         kw = connection._requested
         self.assertEqual(len(kw), 1)
@@ -594,6 +627,7 @@ class Test_ACL(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/testing")
         self.assertEqual(kw[0]["data"], {"acl": []})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], 42)
 
     def test_save_no_acl(self):
         ROLE = "role"
@@ -617,6 +651,7 @@ class Test_ACL(unittest.TestCase):
                 "path": "/testing",
                 "query_params": {"projection": "full"},
                 "data": {"acl": AFTER},
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -648,6 +683,7 @@ class Test_ACL(unittest.TestCase):
                 "path": "/testing",
                 "query_params": {"projection": "full", "userProject": USER_PROJECT},
                 "data": {"acl": new_acl},
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -667,7 +703,7 @@ class Test_ACL(unittest.TestCase):
         acl = self._make_one()
         acl.save_path = "/testing"
         acl.loaded = True
-        acl.save_predefined(PREDEFINED, client=client)
+        acl.save_predefined(PREDEFINED, client=client, timeout=42)
         entries = list(acl)
         self.assertEqual(len(entries), 0)
         kw = connection._requested
@@ -679,6 +715,7 @@ class Test_ACL(unittest.TestCase):
                 "path": "/testing",
                 "query_params": {"projection": "full", "predefinedAcl": PREDEFINED},
                 "data": {"acl": []},
+                "timeout": 42,
             },
         )
 
@@ -705,6 +742,7 @@ class Test_ACL(unittest.TestCase):
                     "predefinedAcl": PREDEFINED_JSON,
                 },
                 "data": {"acl": []},
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -729,6 +767,7 @@ class Test_ACL(unittest.TestCase):
                 "path": "/testing",
                 "query_params": {"projection": "full", "alternate": PREDEFINED},
                 "data": {"acl": []},
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -742,7 +781,7 @@ class Test_ACL(unittest.TestCase):
         acl.save_path = "/testing"
         acl.loaded = True
         acl.entity("allUsers", ROLE1)
-        acl.clear(client=client)
+        acl.clear(client=client, timeout=42)
         self.assertEqual(list(acl), [STICKY])
         kw = connection._requested
         self.assertEqual(len(kw), 1)
@@ -753,6 +792,7 @@ class Test_ACL(unittest.TestCase):
                 "path": "/testing",
                 "query_params": {"projection": "full"},
                 "data": {"acl": []},
+                "timeout": 42,
             },
         )
 
diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py
index e18b1b9..ec8fe75 100644
--- a/tests/unit/test_batch.py
+++ b/tests/unit/test_batch.py
@@ -90,6 +90,12 @@ class TestMIMEApplicationHTTP(unittest.TestCase):
 
 
 class TestBatch(unittest.TestCase):
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     @staticmethod
     def _get_target_class():
         from google.cloud.storage.batch import Batch
@@ -344,7 +350,7 @@ class TestBatch(unittest.TestCase):
             url=expected_url,
             headers=mock.ANY,
             data=mock.ANY,
-            timeout=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
         request_info = self._get_mutlipart_request(http)
@@ -393,8 +399,8 @@ class TestBatch(unittest.TestCase):
         target1 = _MockObject()
         target2 = _MockObject()
 
-        batch._do_request("GET", url, {}, None, target1)
-        batch._do_request("GET", url, {}, None, target2)
+        batch._do_request("GET", url, {}, None, target1, timeout=42)
+        batch._do_request("GET", url, {}, None, target2, timeout=420)
 
         # Make sure futures are not populated.
         self.assertEqual(
@@ -414,7 +420,7 @@ class TestBatch(unittest.TestCase):
             url=expected_url,
             headers=mock.ANY,
             data=mock.ANY,
-            timeout=mock.ANY,
+            timeout=420,  # the last request timeout prevails
         )
 
         _, request_body, _, boundary = self._get_mutlipart_request(http)
diff --git a/tests/unit/test_blob.py b/tests/unit/test_blob.py
index 746e659..1c2b1e9 100644
--- a/tests/unit/test_blob.py
+++ b/tests/unit/test_blob.py
@@ -43,6 +43,12 @@ class Test_Blob(unittest.TestCase):
         blob._properties.update(properties)
         return blob
 
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     def test_ctor_wo_encryption_key(self):
         BLOB_NAME = "blob-name"
         bucket = _Bucket()
@@ -609,7 +615,7 @@ class Test_Blob(unittest.TestCase):
         client = _Client(connection)
         bucket = _Bucket(client)
         blob = self._make_one(NONESUCH, bucket=bucket)
-        self.assertFalse(blob.exists())
+        self.assertFalse(blob.exists(timeout=42))
         self.assertEqual(len(connection._requested), 1)
         self.assertEqual(
             connection._requested[0],
@@ -618,6 +624,7 @@ class Test_Blob(unittest.TestCase):
                 "path": "/b/name/o/{}".format(NONESUCH),
                 "query_params": {"fields": "name"},
                 "_target_object": None,
+                "timeout": 42,
             },
         )
 
@@ -639,6 +646,7 @@ class Test_Blob(unittest.TestCase):
                 "path": "/b/name/o/{}".format(BLOB_NAME),
                 "query_params": {"fields": "name", "userProject": USER_PROJECT},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -660,6 +668,7 @@ class Test_Blob(unittest.TestCase):
                 "path": "/b/name/o/{}".format(BLOB_NAME),
                 "query_params": {"fields": "name", "generation": GENERATION},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -673,7 +682,9 @@ class Test_Blob(unittest.TestCase):
         bucket._blobs[BLOB_NAME] = 1
         blob.delete()
         self.assertFalse(blob.exists())
-        self.assertEqual(bucket._deleted, [(BLOB_NAME, None, None)])
+        self.assertEqual(
+            bucket._deleted, [(BLOB_NAME, None, None, self._get_default_timeout())]
+        )
 
     def test_delete_w_generation(self):
         BLOB_NAME = "blob-name"
@@ -684,9 +695,9 @@ class Test_Blob(unittest.TestCase):
         bucket = _Bucket(client)
         blob = self._make_one(BLOB_NAME, bucket=bucket, generation=GENERATION)
         bucket._blobs[BLOB_NAME] = 1
-        blob.delete()
+        blob.delete(timeout=42)
         self.assertFalse(blob.exists())
-        self.assertEqual(bucket._deleted, [(BLOB_NAME, None, GENERATION)])
+        self.assertEqual(bucket._deleted, [(BLOB_NAME, None, GENERATION, 42)])
 
     def test__get_transport(self):
         client = mock.Mock(spec=[u"_credentials", "_http"])
@@ -1960,7 +1971,7 @@ class Test_Blob(unittest.TestCase):
         bucket = _Bucket(client=client)
         blob = self._make_one(BLOB_NAME, bucket=bucket)
 
-        policy = blob.get_iam_policy()
+        policy = blob.get_iam_policy(timeout=42)
 
         self.assertIsInstance(policy, Policy)
         self.assertEqual(policy.etag, RETURNED["etag"])
@@ -1976,6 +1987,7 @@ class Test_Blob(unittest.TestCase):
                 "path": "%s/iam" % (PATH,),
                 "query_params": {},
                 "_target_object": None,
+                "timeout": 42,
             },
         )
 
@@ -2011,6 +2023,7 @@ class Test_Blob(unittest.TestCase):
                 "path": "%s/iam" % (PATH,),
                 "query_params": {"optionsRequestedPolicyVersion": 3},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -2051,6 +2064,7 @@ class Test_Blob(unittest.TestCase):
                 "path": "%s/iam" % (PATH,),
                 "query_params": {"userProject": USER_PROJECT},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -2087,7 +2101,7 @@ class Test_Blob(unittest.TestCase):
         bucket = _Bucket(client=client)
         blob = self._make_one(BLOB_NAME, bucket=bucket)
 
-        returned = blob.set_iam_policy(policy)
+        returned = blob.set_iam_policy(policy, timeout=42)
 
         self.assertEqual(returned.etag, ETAG)
         self.assertEqual(returned.version, VERSION)
@@ -2098,6 +2112,7 @@ class Test_Blob(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "PUT")
         self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {})
+        self.assertEqual(kw[0]["timeout"], 42)
         sent = kw[0]["data"]
         self.assertEqual(sent["resourceId"], PATH)
         self.assertEqual(len(sent["bindings"]), len(BINDINGS))
@@ -2159,7 +2174,7 @@ class Test_Blob(unittest.TestCase):
         bucket = _Bucket(client=client)
         blob = self._make_one(BLOB_NAME, bucket=bucket)
 
-        allowed = blob.test_iam_permissions(PERMISSIONS)
+        allowed = blob.test_iam_permissions(PERMISSIONS, timeout=42)
 
         self.assertEqual(allowed, ALLOWED)
 
@@ -2168,6 +2183,7 @@ class Test_Blob(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "GET")
         self.assertEqual(kw[0]["path"], "%s/iam/testPermissions" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {"permissions": PERMISSIONS})
+        self.assertEqual(kw[0]["timeout"], 42)
 
     def test_test_iam_permissions_w_user_project(self):
         from google.cloud.storage.iam import STORAGE_OBJECTS_LIST
@@ -2202,6 +2218,7 @@ class Test_Blob(unittest.TestCase):
             kw[0]["query_params"],
             {"permissions": PERMISSIONS, "userProject": USER_PROJECT},
         )
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
     def test_make_public(self):
         from google.cloud.storage.acl import _ACLEntity
@@ -2272,6 +2289,7 @@ class Test_Blob(unittest.TestCase):
                     "destination": {},
                 },
                 "_target_object": destination,
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -2290,7 +2308,7 @@ class Test_Blob(unittest.TestCase):
         destination = self._make_one(DESTINATION, bucket=bucket)
         destination.content_type = "text/plain"
 
-        destination.compose(sources=[source_1, source_2])
+        destination.compose(sources=[source_1, source_2], timeout=42)
 
         self.assertEqual(destination.etag, "DEADBEEF")
 
@@ -2307,6 +2325,7 @@ class Test_Blob(unittest.TestCase):
                     "destination": {"contentType": "text/plain"},
                 },
                 "_target_object": destination,
+                "timeout": 42,
             },
         )
 
@@ -2347,6 +2366,7 @@ class Test_Blob(unittest.TestCase):
                     },
                 },
                 "_target_object": destination,
+                "timeout": self._get_default_timeout(),
             },
         )
 
@@ -2400,7 +2420,7 @@ class Test_Blob(unittest.TestCase):
             DEST_BLOB, bucket=dest_bucket, generation=DEST_GENERATION
         )
 
-        token, rewritten, size = dest_blob.rewrite(source_blob)
+        token, rewritten, size = dest_blob.rewrite(source_blob, timeout=42)
 
         self.assertEqual(token, TOKEN)
         self.assertEqual(rewritten, 33)
@@ -2416,6 +2436,7 @@ class Test_Blob(unittest.TestCase):
             ),
         )
         self.assertEqual(kw["query_params"], {"sourceGeneration": SOURCE_GENERATION})
+        self.assertEqual(kw["timeout"], 42)
 
     def test_rewrite_other_bucket_other_name_no_encryption_partial(self):
         SOURCE_BLOB = "source"
@@ -2455,6 +2476,7 @@ class Test_Blob(unittest.TestCase):
         self.assertEqual(kw[0]["query_params"], {})
         SENT = {}
         self.assertEqual(kw[0]["data"], SENT)
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
         headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
         self.assertNotIn("X-Goog-Copy-Source-Encryption-Algorithm", headers)
@@ -2498,6 +2520,7 @@ class Test_Blob(unittest.TestCase):
         self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
         SENT = {}
         self.assertEqual(kw[0]["data"], SENT)
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
         headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
         self.assertNotIn("X-Goog-Copy-Source-Encryption-Algorithm", headers)
@@ -2545,6 +2568,7 @@ class Test_Blob(unittest.TestCase):
         self.assertEqual(kw[0]["query_params"], {"rewriteToken": TOKEN})
         SENT = {}
         self.assertEqual(kw[0]["data"], SENT)
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
         headers = {key.title(): str(value) for key, value in kw[0]["headers"].items()}
         self.assertEqual(headers["X-Goog-Copy-Source-Encryption-Algorithm"], "AES256")
@@ -2595,6 +2619,7 @@ class Test_Blob(unittest.TestCase):
         self.assertEqual(
             kw[0]["query_params"], {"destinationKmsKeyName": DEST_KMS_RESOURCE}
         )
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
         SENT = {"kmsKeyName": DEST_KMS_RESOURCE}
         self.assertEqual(kw[0]["data"], SENT)
 
@@ -3345,9 +3370,9 @@ class _Bucket(object):
         self.path = "/b/" + name
         self.user_project = user_project
 
-    def delete_blob(self, blob_name, client=None, generation=None):
+    def delete_blob(self, blob_name, client=None, generation=None, timeout=None):
         del self._blobs[blob_name]
-        self._deleted.append((blob_name, client, generation))
+        self._deleted.append((blob_name, client, generation, timeout))
 
 
 class _Client(object):
diff --git a/tests/unit/test_bucket.py b/tests/unit/test_bucket.py
index 68399b3..312fc0f 100644
--- a/tests/unit/test_bucket.py
+++ b/tests/unit/test_bucket.py
@@ -361,6 +361,12 @@ class Test_Bucket(unittest.TestCase):
 
         return Bucket
 
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     def _make_one(self, client=None, name=None, properties=None, user_project=None):
         if client is None:
             connection = _Connection()
@@ -571,12 +577,13 @@ class Test_Bucket(unittest.TestCase):
         BUCKET_NAME = "bucket-name"
         bucket = self._make_one(name=BUCKET_NAME)
         client = _Client(_FakeConnection)
-        self.assertFalse(bucket.exists(client=client))
+        self.assertFalse(bucket.exists(client=client, timeout=42))
         expected_called_kwargs = {
             "method": "GET",
             "path": bucket.path,
             "query_params": {"fields": "name"},
             "_target_object": None,
+            "timeout": 42,
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -603,6 +610,7 @@ class Test_Bucket(unittest.TestCase):
             "path": bucket.path,
             "query_params": {"fields": "name", "userProject": USER_PROJECT},
             "_target_object": None,
+            "timeout": self._get_default_timeout(),
         }
         expected_cw = [((), expected_called_kwargs)]
         self.assertEqual(_FakeConnection._called_with, expected_cw)
@@ -653,6 +661,7 @@ class Test_Bucket(unittest.TestCase):
             query_params={"project": OTHER_PROJECT},
             data=DATA,
             _target_object=bucket,
+            timeout=self._get_default_timeout(),
         )
 
     def test_create_w_explicit_location(self):
@@ -679,6 +688,7 @@ class Test_Bucket(unittest.TestCase):
             data=DATA,
             _target_object=bucket,
             query_params={"project": "PROJECT"},
+            timeout=self._get_default_timeout(),
         )
         self.assertEqual(bucket.location, LOCATION)
 
@@ -693,7 +703,7 @@ class Test_Bucket(unittest.TestCase):
         client._base_connection = connection
 
         bucket = self._make_one(client=client, name=BUCKET_NAME)
-        bucket.create()
+        bucket.create(timeout=42)
 
         connection.api_request.assert_called_once_with(
             method="POST",
@@ -701,6 +711,7 @@ class Test_Bucket(unittest.TestCase):
             query_params={"project": PROJECT},
             data=DATA,
             _target_object=bucket,
+            timeout=42,
         )
 
     def test_create_w_extra_properties(self):
@@ -750,6 +761,7 @@ class Test_Bucket(unittest.TestCase):
             query_params={"project": PROJECT},
             data=DATA,
             _target_object=bucket,
+            timeout=self._get_default_timeout(),
         )
 
     def test_create_w_predefined_acl_invalid(self):
@@ -784,6 +796,7 @@ class Test_Bucket(unittest.TestCase):
         expected_qp = {"project": PROJECT, "predefinedAcl": "publicRead"}
         self.assertEqual(kw["query_params"], expected_qp)
         self.assertEqual(kw["data"], DATA)
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_create_w_predefined_default_object_acl_invalid(self):
         from google.cloud.storage.client import Client
@@ -817,6 +830,7 @@ class Test_Bucket(unittest.TestCase):
         expected_qp = {"project": PROJECT, "predefinedDefaultObjectAcl": "publicRead"}
         self.assertEqual(kw["query_params"], expected_qp)
         self.assertEqual(kw["data"], DATA)
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_acl_property(self):
         from google.cloud.storage.acl import BucketACL
@@ -849,11 +863,12 @@ class Test_Bucket(unittest.TestCase):
         connection = _Connection()
         client = _Client(connection)
         bucket = self._make_one(name=NAME)
-        result = bucket.get_blob(NONESUCH, client=client)
+        result = bucket.get_blob(NONESUCH, client=client, timeout=42)
         self.assertIsNone(result)
         kw, = connection._requested
         self.assertEqual(kw["method"], "GET")
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, NONESUCH))
+        self.assertEqual(kw["timeout"], 42)
 
     def test_get_blob_hit_w_user_project(self):
         NAME = "name"
@@ -870,6 +885,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "GET")
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw["query_params"], expected_qp)
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_get_blob_hit_w_generation(self):
         NAME = "name"
@@ -887,6 +903,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "GET")
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw["query_params"], expected_qp)
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_get_blob_hit_with_kwargs(self):
         from google.cloud.storage.blob import _get_encryption_headers
@@ -908,6 +925,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "GET")
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw["headers"], _get_encryption_headers(KEY))
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
         self.assertEqual(blob.chunk_size, CHUNK_SIZE)
         self.assertEqual(blob._encryption_key, KEY)
 
@@ -923,6 +941,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "GET")
         self.assertEqual(kw["path"], "/b/%s/o" % NAME)
         self.assertEqual(kw["query_params"], {"projection": "noAcl"})
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_list_blobs_w_all_arguments_and_user_project(self):
         NAME = "name"
@@ -956,6 +975,7 @@ class Test_Bucket(unittest.TestCase):
             projection=PROJECTION,
             fields=FIELDS,
             client=client,
+            timeout=42,
         )
         blobs = list(iterator)
         self.assertEqual(blobs, [])
@@ -963,6 +983,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "GET")
         self.assertEqual(kw["path"], "/b/%s/o" % NAME)
         self.assertEqual(kw["query_params"], EXPECTED)
+        self.assertEqual(kw["timeout"], 42)
 
     def test_list_notifications(self):
         from google.cloud.storage.notification import BucketNotification
@@ -996,7 +1017,10 @@ class Test_Bucket(unittest.TestCase):
         client = _Client(connection)
         bucket = self._make_one(client=client, name=NAME)
 
-        notifications = list(bucket.list_notifications())
+        notifications = list(bucket.list_notifications(timeout=42))
+
+        req_args = client._connection._requested[0]
+        self.assertEqual(req_args.get("timeout"), 42)
 
         self.assertEqual(len(notifications), len(resources))
         for notification, resource, topic_ref in zip(
@@ -1033,6 +1057,7 @@ class Test_Bucket(unittest.TestCase):
                 "path": bucket.path,
                 "query_params": {},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1045,7 +1070,7 @@ class Test_Bucket(unittest.TestCase):
         connection._delete_bucket = True
         client = _Client(connection)
         bucket = self._make_one(client=client, name=NAME, user_project=USER_PROJECT)
-        result = bucket.delete(force=True)
+        result = bucket.delete(force=True, timeout=42)
         self.assertIsNone(result)
         expected_cw = [
             {
@@ -1053,6 +1078,7 @@ class Test_Bucket(unittest.TestCase):
                 "path": bucket.path,
                 "_target_object": None,
                 "query_params": {"userProject": USER_PROJECT},
+                "timeout": 42,
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1075,6 +1101,7 @@ class Test_Bucket(unittest.TestCase):
                 "path": bucket.path,
                 "query_params": {},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1096,6 +1123,7 @@ class Test_Bucket(unittest.TestCase):
                 "path": bucket.path,
                 "query_params": {},
                 "_target_object": None,
+                "timeout": self._get_default_timeout(),
             }
         ]
         self.assertEqual(connection._deleted_buckets, expected_cw)
@@ -1128,6 +1156,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "DELETE")
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, NONESUCH))
         self.assertEqual(kw["query_params"], {})
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_delete_blob_hit_with_user_project(self):
         NAME = "name"
@@ -1136,12 +1165,13 @@ class Test_Bucket(unittest.TestCase):
         connection = _Connection({})
         client = _Client(connection)
         bucket = self._make_one(client=client, name=NAME, user_project=USER_PROJECT)
-        result = bucket.delete_blob(BLOB_NAME)
+        result = bucket.delete_blob(BLOB_NAME, timeout=42)
         self.assertIsNone(result)
         kw, = connection._requested
         self.assertEqual(kw["method"], "DELETE")
         self.assertEqual(kw["path"], "/b/%s/o/%s" % (NAME, BLOB_NAME))
         self.assertEqual(kw["query_params"], {"userProject": USER_PROJECT})
+        self.assertEqual(kw["timeout"], 42)
 
     def test_delete_blob_hit_with_generation(self):
         NAME = "name"
@@ -1156,6 +1186,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "DELETE")
         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())
 
     def test_delete_blobs_empty(self):
         NAME = "name"
@@ -1172,12 +1203,13 @@ class Test_Bucket(unittest.TestCase):
         connection = _Connection({})
         client = _Client(connection)
         bucket = self._make_one(client=client, name=NAME, user_project=USER_PROJECT)
-        bucket.delete_blobs([BLOB_NAME])
+        bucket.delete_blobs([BLOB_NAME], timeout=42)
         kw = connection._requested
         self.assertEqual(len(kw), 1)
         self.assertEqual(kw[0]["method"], "DELETE")
         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)
 
     def test_delete_blobs_miss_no_on_error(self):
         from google.cloud.exceptions import NotFound
@@ -1193,8 +1225,10 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(len(kw), 2)
         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[1]["method"], "DELETE")
         self.assertEqual(kw[1]["path"], "/b/%s/o/%s" % (NAME, NONESUCH))
+        self.assertEqual(kw[1]["timeout"], self._get_default_timeout())
 
     def test_delete_blobs_miss_w_on_error(self):
         NAME = "name"
@@ -1210,8 +1244,10 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(len(kw), 2)
         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[1]["method"], "DELETE")
         self.assertEqual(kw[1]["path"], "/b/%s/o/%s" % (NAME, NONESUCH))
+        self.assertEqual(kw[1]["timeout"], self._get_default_timeout())
 
     @staticmethod
     def _make_blob(bucket_name, blob_name):
@@ -1232,7 +1268,7 @@ class Test_Bucket(unittest.TestCase):
         dest = self._make_one(client=client, name=DEST)
         blob = self._make_blob(SOURCE, BLOB_NAME)
 
-        new_blob = source.copy_blob(blob, dest)
+        new_blob = source.copy_blob(blob, dest, timeout=42)
 
         self.assertIs(new_blob.bucket, dest)
         self.assertEqual(new_blob.name, BLOB_NAME)
@@ -1244,6 +1280,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "POST")
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {})
+        self.assertEqual(kw["timeout"], 42)
 
     def test_copy_blobs_source_generation(self):
         SOURCE = "source"
@@ -1269,6 +1306,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "POST")
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {"sourceGeneration": GENERATION})
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_copy_blobs_preserve_acl(self):
         from google.cloud.storage.acl import ObjectACL
@@ -1301,10 +1339,12 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw1["method"], "POST")
         self.assertEqual(kw1["path"], COPY_PATH)
         self.assertEqual(kw1["query_params"], {})
+        self.assertEqual(kw1["timeout"], self._get_default_timeout())
 
         self.assertEqual(kw2["method"], "PATCH")
         self.assertEqual(kw2["path"], NEW_BLOB_PATH)
         self.assertEqual(kw2["query_params"], {"projection": "full"})
+        self.assertEqual(kw2["timeout"], self._get_default_timeout())
 
     def test_copy_blobs_w_name_and_user_project(self):
         SOURCE = "source"
@@ -1330,6 +1370,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "POST")
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {"userProject": USER_PROJECT})
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_rename_blob(self):
         BUCKET_NAME = "BUCKET_NAME"
@@ -1341,7 +1382,9 @@ class Test_Bucket(unittest.TestCase):
         bucket = self._make_one(client=client, name=BUCKET_NAME)
         blob = self._make_blob(BUCKET_NAME, BLOB_NAME)
 
-        renamed_blob = bucket.rename_blob(blob, NEW_BLOB_NAME, client=client)
+        renamed_blob = bucket.rename_blob(
+            blob, NEW_BLOB_NAME, client=client, timeout=42
+        )
 
         self.assertIs(renamed_blob.bucket, bucket)
         self.assertEqual(renamed_blob.name, NEW_BLOB_NAME)
@@ -1353,8 +1396,9 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "POST")
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {})
+        self.assertEqual(kw["timeout"], 42)
 
-        blob.delete.assert_called_once_with(client)
+        blob.delete.assert_called_once_with(client=client, timeout=42)
 
     def test_rename_blob_to_itself(self):
         BUCKET_NAME = "BUCKET_NAME"
@@ -1377,6 +1421,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw["method"], "POST")
         self.assertEqual(kw["path"], COPY_PATH)
         self.assertEqual(kw["query_params"], {})
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
         blob.delete.assert_not_called()
 
@@ -1680,13 +1725,15 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(len(kwargs["data"]["labels"]), 2)
         self.assertEqual(kwargs["data"]["labels"]["color"], "red")
         self.assertIsNone(kwargs["data"]["labels"]["flavor"])
+        self.assertEqual(kwargs["timeout"], self._get_default_timeout())
 
         # A second patch call should be a no-op for labels.
         client._connection.api_request.reset_mock()
-        bucket.patch(client=client)
+        bucket.patch(client=client, timeout=42)
         client._connection.api_request.assert_called()
         _, _, kwargs = client._connection.api_request.mock_calls[0]
         self.assertNotIn("labels", kwargs["data"])
+        self.assertEqual(kwargs["timeout"], 42)
 
     def test_location_type_getter_unset(self):
         bucket = self._make_one()
@@ -2047,7 +2094,7 @@ class Test_Bucket(unittest.TestCase):
         client = _Client(connection, None)
         bucket = self._make_one(client=client, name=NAME)
 
-        policy = bucket.get_iam_policy()
+        policy = bucket.get_iam_policy(timeout=42)
 
         self.assertIsInstance(policy, Policy)
         self.assertEqual(policy.etag, RETURNED["etag"])
@@ -2059,6 +2106,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "GET")
         self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {})
+        self.assertEqual(kw[0]["timeout"], 42)
 
     def test_get_iam_policy_w_user_project(self):
         from google.api_core.iam import Policy
@@ -2091,6 +2139,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "GET")
         self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
     def test_get_iam_policy_w_requested_policy_version(self):
         from google.cloud.storage.iam import STORAGE_OWNER_ROLE
@@ -2120,6 +2169,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "GET")
         self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {"optionsRequestedPolicyVersion": 3})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
     def test_set_iam_policy(self):
         import operator
@@ -2152,7 +2202,7 @@ class Test_Bucket(unittest.TestCase):
         client = _Client(connection, None)
         bucket = self._make_one(client=client, name=NAME)
 
-        returned = bucket.set_iam_policy(policy)
+        returned = bucket.set_iam_policy(policy, timeout=42)
 
         self.assertEqual(returned.etag, ETAG)
         self.assertEqual(returned.version, VERSION)
@@ -2163,6 +2213,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "PUT")
         self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {})
+        self.assertEqual(kw[0]["timeout"], 42)
         sent = kw[0]["data"]
         self.assertEqual(sent["resourceId"], PATH)
         self.assertEqual(len(sent["bindings"]), len(BINDINGS))
@@ -2216,6 +2267,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "PUT")
         self.assertEqual(kw[0]["path"], "%s/iam" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {"userProject": USER_PROJECT})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
         sent = kw[0]["data"]
         self.assertEqual(sent["resourceId"], PATH)
         self.assertEqual(len(sent["bindings"]), len(BINDINGS))
@@ -2244,7 +2296,7 @@ class Test_Bucket(unittest.TestCase):
         client = _Client(connection, None)
         bucket = self._make_one(client=client, name=NAME)
 
-        allowed = bucket.test_iam_permissions(PERMISSIONS)
+        allowed = bucket.test_iam_permissions(PERMISSIONS, timeout=42)
 
         self.assertEqual(allowed, ALLOWED)
 
@@ -2253,6 +2305,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["method"], "GET")
         self.assertEqual(kw[0]["path"], "%s/iam/testPermissions" % (PATH,))
         self.assertEqual(kw[0]["query_params"], {"permissions": PERMISSIONS})
+        self.assertEqual(kw[0]["timeout"], 42)
 
     def test_test_iam_permissions_w_user_project(self):
         from google.cloud.storage.iam import STORAGE_OBJECTS_LIST
@@ -2285,6 +2338,7 @@ class Test_Bucket(unittest.TestCase):
             kw[0]["query_params"],
             {"permissions": PERMISSIONS, "userProject": USER_PROJECT},
         )
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
     def test_make_public_defaults(self):
         from google.cloud.storage.acl import _ACLEntity
@@ -2306,6 +2360,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[0]["data"], {"acl": after["acl"]})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
     def _make_public_w_future_helper(self, default_object_acl_loaded=True):
         from google.cloud.storage.acl import _ACLEntity
@@ -2335,6 +2390,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[0]["data"], {"acl": permissive})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
         if not default_object_acl_loaded:
             self.assertEqual(kw[1]["method"], "GET")
             self.assertEqual(kw[1]["path"], "/b/%s/defaultObjectAcl" % NAME)
@@ -2343,6 +2399,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[-1]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[-1]["data"], {"defaultObjectAcl": permissive})
         self.assertEqual(kw[-1]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[-1]["timeout"], self._get_default_timeout())
 
     def test_make_public_w_future(self):
         self._make_public_w_future_helper(default_object_acl_loaded=True)
@@ -2373,8 +2430,10 @@ class Test_Bucket(unittest.TestCase):
             def grant_read(self):
                 self._granted = True
 
-            def save(self, client=None):
-                _saved.append((self._bucket, self._name, self._granted, client))
+            def save(self, client=None, timeout=None):
+                _saved.append(
+                    (self._bucket, self._name, self._granted, client, timeout)
+                )
 
         def item_to_blob(self, item):
             return _Blob(self.bucket, item["name"])
@@ -2390,22 +2449,24 @@ class Test_Bucket(unittest.TestCase):
         bucket.default_object_acl.loaded = True
 
         with mock.patch("google.cloud.storage.bucket._item_to_blob", new=item_to_blob):
-            bucket.make_public(recursive=True)
+            bucket.make_public(recursive=True, timeout=42)
         self.assertEqual(list(bucket.acl), permissive)
         self.assertEqual(list(bucket.default_object_acl), [])
-        self.assertEqual(_saved, [(bucket, BLOB_NAME, True, None)])
+        self.assertEqual(_saved, [(bucket, BLOB_NAME, True, None, 42)])
         kw = connection._requested
         self.assertEqual(len(kw), 2)
         self.assertEqual(kw[0]["method"], "PATCH")
         self.assertEqual(kw[0]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[0]["data"], {"acl": permissive})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], 42)
         self.assertEqual(kw[1]["method"], "GET")
         self.assertEqual(kw[1]["path"], "/b/%s/o" % NAME)
         max_results = bucket._MAX_OBJECTS_FOR_ITERATION + 1
         self.assertEqual(
             kw[1]["query_params"], {"maxResults": max_results, "projection": "full"}
         )
+        self.assertEqual(kw[1]["timeout"], 42)
 
     def test_make_public_recursive_too_many(self):
         from google.cloud.storage.acl import _ACLEntity
@@ -2445,6 +2506,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[0]["data"], {"acl": after["acl"]})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
 
     def _make_private_w_future_helper(self, default_object_acl_loaded=True):
         NAME = "name"
@@ -2472,6 +2534,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[0]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[0]["data"], {"acl": no_permissions})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], self._get_default_timeout())
         if not default_object_acl_loaded:
             self.assertEqual(kw[1]["method"], "GET")
             self.assertEqual(kw[1]["path"], "/b/%s/defaultObjectAcl" % NAME)
@@ -2480,6 +2543,7 @@ class Test_Bucket(unittest.TestCase):
         self.assertEqual(kw[-1]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[-1]["data"], {"defaultObjectAcl": no_permissions})
         self.assertEqual(kw[-1]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[-1]["timeout"], self._get_default_timeout())
 
     def test_make_private_w_future(self):
         self._make_private_w_future_helper(default_object_acl_loaded=True)
@@ -2508,8 +2572,10 @@ class Test_Bucket(unittest.TestCase):
             def revoke_read(self):
                 self._granted = False
 
-            def save(self, client=None):
-                _saved.append((self._bucket, self._name, self._granted, client))
+            def save(self, client=None, timeout=None):
+                _saved.append(
+                    (self._bucket, self._name, self._granted, client, timeout)
+                )
 
         def item_to_blob(self, item):
             return _Blob(self.bucket, item["name"])
@@ -2525,22 +2591,24 @@ class Test_Bucket(unittest.TestCase):
         bucket.default_object_acl.loaded = True
 
         with mock.patch("google.cloud.storage.bucket._item_to_blob", new=item_to_blob):
-            bucket.make_private(recursive=True)
+            bucket.make_private(recursive=True, timeout=42)
         self.assertEqual(list(bucket.acl), no_permissions)
         self.assertEqual(list(bucket.default_object_acl), [])
-        self.assertEqual(_saved, [(bucket, BLOB_NAME, False, None)])
+        self.assertEqual(_saved, [(bucket, BLOB_NAME, False, None, 42)])
         kw = connection._requested
         self.assertEqual(len(kw), 2)
         self.assertEqual(kw[0]["method"], "PATCH")
         self.assertEqual(kw[0]["path"], "/b/%s" % NAME)
         self.assertEqual(kw[0]["data"], {"acl": no_permissions})
         self.assertEqual(kw[0]["query_params"], {"projection": "full"})
+        self.assertEqual(kw[0]["timeout"], 42)
         self.assertEqual(kw[1]["method"], "GET")
         self.assertEqual(kw[1]["path"], "/b/%s/o" % NAME)
         max_results = bucket._MAX_OBJECTS_FOR_ITERATION + 1
         self.assertEqual(
             kw[1]["query_params"], {"maxResults": max_results, "projection": "full"}
         )
+        self.assertEqual(kw[1]["timeout"], 42)
 
     def test_make_private_recursive_too_many(self):
         NO_PERMISSIONS = []
@@ -2778,12 +2846,13 @@ class Test_Bucket(unittest.TestCase):
             "retentionPeriod": 86400 * 100,  # 100 days
         }
 
-        bucket.lock_retention_policy()
+        bucket.lock_retention_policy(timeout=42)
 
         kw, = connection._requested
         self.assertEqual(kw["method"], "POST")
         self.assertEqual(kw["path"], "/b/{}/lockRetentionPolicy".format(name))
         self.assertEqual(kw["query_params"], {"ifMetagenerationMatch": 1234})
+        self.assertEqual(kw["timeout"], 42)
 
     def test_lock_retention_policy_w_user_project(self):
         name = "name"
@@ -2817,6 +2886,7 @@ class Test_Bucket(unittest.TestCase):
             kw["query_params"],
             {"ifMetagenerationMatch": 1234, "userProject": user_project},
         )
+        self.assertEqual(kw["timeout"], self._get_default_timeout())
 
     def test_generate_signed_url_w_invalid_version(self):
         expiration = "2014-10-16T20:34:37.000Z"
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index f3c090e..b3e5874 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -68,6 +68,12 @@ class TestClient(unittest.TestCase):
 
         return Client
 
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     def _make_one(self, *args, **kw):
         return self._get_target_class()(*args, **kw)
 
@@ -259,7 +265,7 @@ class TestClient(unittest.TestCase):
         http = _make_requests_session([_make_json_response(RESOURCE)])
         client._http_internal = http
 
-        service_account_email = client.get_service_account_email()
+        service_account_email = client.get_service_account_email(timeout=42)
 
         self.assertEqual(service_account_email, EMAIL)
         URI = "/".join(
@@ -271,7 +277,7 @@ class TestClient(unittest.TestCase):
             ]
         )
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="GET", url=URI, data=None, headers=mock.ANY, timeout=42
         )
 
     def test_get_service_account_email_w_project(self):
@@ -297,7 +303,11 @@ class TestClient(unittest.TestCase):
             ]
         )
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=URI,
+            data=None,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_bucket(self):
@@ -363,10 +373,10 @@ class TestClient(unittest.TestCase):
         client._http_internal = http
 
         with self.assertRaises(NotFound):
-            client.get_bucket(NONESUCH)
+            client.get_bucket(NONESUCH, timeout=42)
 
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=42
         )
 
     def test_get_bucket_with_string_hit(self):
@@ -396,7 +406,11 @@ class TestClient(unittest.TestCase):
         self.assertIsInstance(bucket, Bucket)
         self.assertEqual(bucket.name, BUCKET_NAME)
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=URI,
+            data=mock.ANY,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_get_bucket_with_object_miss(self):
@@ -427,7 +441,11 @@ class TestClient(unittest.TestCase):
             client.get_bucket(bucket_obj)
 
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=URI,
+            data=mock.ANY,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_get_bucket_with_object_hit(self):
@@ -458,7 +476,11 @@ class TestClient(unittest.TestCase):
         self.assertIsInstance(bucket, Bucket)
         self.assertEqual(bucket.name, bucket_name)
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=URI,
+            data=mock.ANY,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_lookup_bucket_miss(self):
@@ -481,11 +503,11 @@ class TestClient(unittest.TestCase):
         )
         client._http_internal = http
 
-        bucket = client.lookup_bucket(NONESUCH)
+        bucket = client.lookup_bucket(NONESUCH, timeout=42)
 
         self.assertIsNone(bucket)
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=42
         )
 
     def test_lookup_bucket_hit(self):
@@ -514,7 +536,11 @@ class TestClient(unittest.TestCase):
         self.assertIsInstance(bucket, Bucket)
         self.assertEqual(bucket.name, BUCKET_NAME)
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=mock.ANY, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=URI,
+            data=mock.ANY,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_create_bucket_w_missing_client_project(self):
@@ -550,6 +576,7 @@ class TestClient(unittest.TestCase):
             query_params={"project": other_project, "userProject": user_project},
             data=data,
             _target_object=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_create_bucket_w_predefined_acl_invalid(self):
@@ -570,7 +597,9 @@ class TestClient(unittest.TestCase):
         client = self._make_one(project=project, credentials=credentials)
         connection = _make_connection(data)
         client._base_connection = connection
-        bucket = client.create_bucket(bucket_name, predefined_acl="publicRead")
+        bucket = client.create_bucket(
+            bucket_name, predefined_acl="publicRead", timeout=42
+        )
 
         connection.api_request.assert_called_once_with(
             method="POST",
@@ -578,6 +607,7 @@ class TestClient(unittest.TestCase):
             query_params={"project": project, "predefinedAcl": "publicRead"},
             data=data,
             _target_object=bucket,
+            timeout=42,
         )
 
     def test_create_bucket_w_predefined_default_object_acl_invalid(self):
@@ -612,6 +642,7 @@ class TestClient(unittest.TestCase):
             },
             data=data,
             _target_object=bucket,
+            timeout=self._get_default_timeout(),
         )
 
     def test_create_bucket_w_explicit_location(self):
@@ -636,6 +667,7 @@ class TestClient(unittest.TestCase):
             data=data,
             _target_object=bucket,
             query_params={"project": project},
+            timeout=self._get_default_timeout(),
         )
         self.assertEqual(bucket.location, location)
 
@@ -772,6 +804,7 @@ class TestClient(unittest.TestCase):
                 method="GET",
                 path="/b/%s/o" % BUCKET_NAME,
                 query_params={"projection": "noAcl"},
+                timeout=self._get_default_timeout(),
             )
 
     def test_list_blobs_w_all_arguments_and_user_project(self):
@@ -817,12 +850,16 @@ class TestClient(unittest.TestCase):
                 versions=VERSIONS,
                 projection=PROJECTION,
                 fields=FIELDS,
+                timeout=42,
             )
             blobs = list(iterator)
 
             self.assertEqual(blobs, [])
             connection.api_request.assert_called_once_with(
-                method="GET", path="/b/%s/o" % BUCKET_NAME, query_params=EXPECTED
+                method="GET",
+                path="/b/%s/o" % BUCKET_NAME,
+                query_params=EXPECTED,
+                timeout=42,
             )
 
     def test_list_buckets_wo_project(self):
@@ -930,7 +967,7 @@ class TestClient(unittest.TestCase):
             url=mock.ANY,
             data=mock.ANY,
             headers=mock.ANY,
-            timeout=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_list_buckets_all_arguments(self):
@@ -956,15 +993,12 @@ class TestClient(unittest.TestCase):
             prefix=PREFIX,
             projection=PROJECTION,
             fields=FIELDS,
+            timeout=42,
         )
         buckets = list(iterator)
         self.assertEqual(buckets, [])
         http.request.assert_called_once_with(
-            method="GET",
-            url=mock.ANY,
-            data=mock.ANY,
-            headers=mock.ANY,
-            timeout=mock.ANY,
+            method="GET", url=mock.ANY, data=mock.ANY, headers=mock.ANY, timeout=42
         )
 
         requested_url = http.request.mock_calls[0][2]["url"]
@@ -1024,7 +1058,9 @@ class TestClient(unittest.TestCase):
         self.assertIsInstance(bucket, Bucket)
         self.assertEqual(bucket.name, blob_name)
 
-    def _create_hmac_key_helper(self, explicit_project=None, user_project=None):
+    def _create_hmac_key_helper(
+        self, explicit_project=None, user_project=None, timeout=None
+    ):
         import datetime
         from pytz import UTC
         from six.moves.urllib.parse import urlencode
@@ -1069,6 +1105,10 @@ class TestClient(unittest.TestCase):
         if user_project is not None:
             kwargs["user_project"] = user_project
 
+        if timeout is None:
+            timeout = self._get_default_timeout()
+        kwargs["timeout"] = timeout
+
         metadata, secret = client.create_hmac_key(service_account_email=EMAIL, **kwargs)
 
         self.assertIsInstance(metadata, HMACKeyMetadata)
@@ -1093,7 +1133,7 @@ class TestClient(unittest.TestCase):
 
         FULL_URI = "{}?{}".format(URI, urlencode(qs_params))
         http.request.assert_called_once_with(
-            method="POST", url=FULL_URI, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="POST", url=FULL_URI, data=None, headers=mock.ANY, timeout=timeout
         )
 
     def test_create_hmac_key_defaults(self):
@@ -1103,7 +1143,7 @@ class TestClient(unittest.TestCase):
         self._create_hmac_key_helper(explicit_project="other-project-456")
 
     def test_create_hmac_key_user_project(self):
-        self._create_hmac_key_helper(user_project="billed-project")
+        self._create_hmac_key_helper(user_project="billed-project", timeout=42)
 
     def test_list_hmac_keys_defaults_empty(self):
         PROJECT = "PROJECT"
@@ -1128,7 +1168,11 @@ class TestClient(unittest.TestCase):
             ]
         )
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=URI,
+            data=None,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
 
     def test_list_hmac_keys_explicit_non_empty(self):
@@ -1165,6 +1209,7 @@ class TestClient(unittest.TestCase):
                 show_deleted_keys=True,
                 project_id=OTHER_PROJECT,
                 user_project=USER_PROJECT,
+                timeout=42,
             )
         )
 
@@ -1192,7 +1237,7 @@ class TestClient(unittest.TestCase):
             "userProject": USER_PROJECT,
         }
         http.request.assert_called_once_with(
-            method="GET", url=mock.ANY, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="GET", url=mock.ANY, data=None, headers=mock.ANY, timeout=42
         )
         kwargs = http.request.mock_calls[0].kwargs
         uri = kwargs["url"]
@@ -1220,7 +1265,7 @@ class TestClient(unittest.TestCase):
         http = _make_requests_session([_make_json_response(resource)])
         client._http_internal = http
 
-        metadata = client.get_hmac_key_metadata(ACCESS_ID)
+        metadata = client.get_hmac_key_metadata(ACCESS_ID, timeout=42)
 
         self.assertIsInstance(metadata, HMACKeyMetadata)
         self.assertIs(metadata._client, client)
@@ -1239,7 +1284,7 @@ class TestClient(unittest.TestCase):
             ]
         )
         http.request.assert_called_once_with(
-            method="GET", url=URI, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="GET", url=URI, data=None, headers=mock.ANY, timeout=42
         )
 
     def test_get_hmac_key_metadata_w_project(self):
@@ -1289,5 +1334,9 @@ class TestClient(unittest.TestCase):
         FULL_URI = "{}?{}".format(URI, urlencode(qs_params))
 
         http.request.assert_called_once_with(
-            method="GET", url=FULL_URI, data=None, headers=mock.ANY, timeout=mock.ANY
+            method="GET",
+            url=FULL_URI,
+            data=None,
+            headers=mock.ANY,
+            timeout=self._get_default_timeout(),
         )
diff --git a/tests/unit/test_hmac_key.py b/tests/unit/test_hmac_key.py
index 138742d..a142939 100644
--- a/tests/unit/test_hmac_key.py
+++ b/tests/unit/test_hmac_key.py
@@ -18,6 +18,12 @@ import mock
 
 
 class TestHMACKeyMetadata(unittest.TestCase):
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     @staticmethod
     def _get_target_class():
         from google.cloud.storage.hmac_key import HMACKeyMetadata
@@ -219,12 +225,17 @@ class TestHMACKeyMetadata(unittest.TestCase):
         metadata = self._make_one(client)
         metadata._properties["accessId"] = access_id
 
-        self.assertFalse(metadata.exists())
+        self.assertFalse(metadata.exists(timeout=42))
 
         expected_path = "/projects/{}/hmacKeys/{}".format(
             client.DEFAULT_PROJECT, access_id
         )
-        expected_kwargs = {"method": "GET", "path": expected_path, "query_params": {}}
+        expected_kwargs = {
+            "method": "GET",
+            "path": expected_path,
+            "query_params": {},
+            "timeout": 42,
+        }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
     def test_exists_hit_w_project_set(self):
@@ -251,6 +262,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "method": "GET",
             "path": expected_path,
             "query_params": {"userProject": user_project},
+            "timeout": self._get_default_timeout(),
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -265,12 +277,17 @@ class TestHMACKeyMetadata(unittest.TestCase):
         metadata._properties["accessId"] = access_id
 
         with self.assertRaises(NotFound):
-            metadata.reload()
+            metadata.reload(timeout=42)
 
         expected_path = "/projects/{}/hmacKeys/{}".format(
             client.DEFAULT_PROJECT, access_id
         )
-        expected_kwargs = {"method": "GET", "path": expected_path, "query_params": {}}
+        expected_kwargs = {
+            "method": "GET",
+            "path": expected_path,
+            "query_params": {},
+            "timeout": 42,
+        }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
     def test_reload_hit_w_project_set(self):
@@ -299,6 +316,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "method": "GET",
             "path": expected_path,
             "query_params": {"userProject": user_project},
+            "timeout": self._get_default_timeout(),
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -314,7 +332,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
         metadata.state = "INACTIVE"
 
         with self.assertRaises(NotFound):
-            metadata.update()
+            metadata.update(timeout=42)
 
         expected_path = "/projects/{}/hmacKeys/{}".format(
             client.DEFAULT_PROJECT, access_id
@@ -324,6 +342,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "data": {"state": "INACTIVE"},
             "query_params": {},
+            "timeout": 42,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -356,6 +375,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "path": expected_path,
             "data": {"state": "ACTIVE"},
             "query_params": {"userProject": user_project},
+            "timeout": self._get_default_timeout(),
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -379,7 +399,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
         metadata.state = "INACTIVE"
 
         with self.assertRaises(NotFound):
-            metadata.delete()
+            metadata.delete(timeout=42)
 
         expected_path = "/projects/{}/hmacKeys/{}".format(
             client.DEFAULT_PROJECT, access_id
@@ -388,6 +408,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "method": "DELETE",
             "path": expected_path,
             "query_params": {},
+            "timeout": 42,
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
@@ -410,6 +431,7 @@ class TestHMACKeyMetadata(unittest.TestCase):
             "method": "DELETE",
             "path": expected_path,
             "query_params": {"userProject": user_project},
+            "timeout": self._get_default_timeout(),
         }
         connection.api_request.assert_called_once_with(**expected_kwargs)
 
diff --git a/tests/unit/test_notification.py b/tests/unit/test_notification.py
index 29b376b..f056701 100644
--- a/tests/unit/test_notification.py
+++ b/tests/unit/test_notification.py
@@ -51,6 +51,12 @@ class TestBucketNotification(unittest.TestCase):
 
         return JSON_API_V1_PAYLOAD_FORMAT
 
+    @staticmethod
+    def _get_default_timeout():
+        from google.cloud.storage.constants import _DEFAULT_TIMEOUT
+
+        return _DEFAULT_TIMEOUT
+
     @staticmethod
     def _get_target_class():
         from google.cloud.storage.notification import BucketNotification
@@ -258,7 +264,11 @@ class TestBucketNotification(unittest.TestCase):
 
         data = {"topic": self.TOPIC_REF, "payload_format": NONE_PAYLOAD_FORMAT}
         api_request.assert_called_once_with(
-            method="POST", path=self.CREATE_PATH, query_params={}, data=data
+            method="POST",
+            path=self.CREATE_PATH,
+            query_params={},
+            data=data,
+            timeout=self._get_default_timeout(),
         )
 
     def test_create_w_explicit_client(self):
@@ -287,7 +297,7 @@ class TestBucketNotification(unittest.TestCase):
             "selfLink": self.SELF_LINK,
         }
 
-        notification.create(client=alt_client)
+        notification.create(client=alt_client, timeout=42)
 
         self.assertEqual(notification.custom_attributes, self.CUSTOM_ATTRIBUTES)
         self.assertEqual(notification.event_types, self.event_types())
@@ -309,6 +319,7 @@ class TestBucketNotification(unittest.TestCase):
             path=self.CREATE_PATH,
             query_params={"userProject": USER_PROJECT},
             data=data,
+            timeout=42,
         )
 
     def test_exists_wo_notification_id(self):
@@ -329,10 +340,10 @@ class TestBucketNotification(unittest.TestCase):
         api_request = client._connection.api_request
         api_request.side_effect = NotFound("testing")
 
-        self.assertFalse(notification.exists())
+        self.assertFalse(notification.exists(timeout=42))
 
         api_request.assert_called_once_with(
-            method="GET", path=self.NOTIFICATION_PATH, query_params={}
+            method="GET", path=self.NOTIFICATION_PATH, query_params={}, timeout=42
         )
 
     def test_exists_hit(self):
@@ -355,6 +366,7 @@ class TestBucketNotification(unittest.TestCase):
             method="GET",
             path=self.NOTIFICATION_PATH,
             query_params={"userProject": USER_PROJECT},
+            timeout=self._get_default_timeout(),
         )
 
     def test_reload_wo_notification_id(self):
@@ -376,10 +388,10 @@ class TestBucketNotification(unittest.TestCase):
         api_request.side_effect = NotFound("testing")
 
         with self.assertRaises(NotFound):
-            notification.reload()
+            notification.reload(timeout=42)
 
         api_request.assert_called_once_with(
-            method="GET", path=self.NOTIFICATION_PATH, query_params={}
+            method="GET", path=self.NOTIFICATION_PATH, query_params={}, timeout=42
         )
 
     def test_reload_hit(self):
@@ -412,6 +424,7 @@ class TestBucketNotification(unittest.TestCase):
             method="GET",
             path=self.NOTIFICATION_PATH,
             query_params={"userProject": USER_PROJECT},
+            timeout=self._get_default_timeout(),
         )
 
     def test_delete_wo_notification_id(self):
@@ -433,10 +446,10 @@ class TestBucketNotification(unittest.TestCase):
         api_request.side_effect = NotFound("testing")
 
         with self.assertRaises(NotFound):
-            notification.delete()
+            notification.delete(timeout=42)
 
         api_request.assert_called_once_with(
-            method="DELETE", path=self.NOTIFICATION_PATH, query_params={}
+            method="DELETE", path=self.NOTIFICATION_PATH, query_params={}, timeout=42
         )
 
     def test_delete_hit(self):
@@ -454,6 +467,7 @@ class TestBucketNotification(unittest.TestCase):
             method="DELETE",
             path=self.NOTIFICATION_PATH,
             query_params={"userProject": USER_PROJECT},
+            timeout=self._get_default_timeout(),
         )