From cd0d443ac21093f14c616694d68243725486b992 Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Tue, 18 Jul 2017 13:57:12 -0700
Subject: [PATCH 1/8] Add basic future interface to bigquery

---
 bigquery/google/cloud/bigquery/job.py | 58 ++++++++++++++++++++++++++-
 1 file changed, 56 insertions(+), 2 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 4f791bdbea0c..19a616c1dd3f 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -14,8 +14,11 @@
 
 """Define API Jobs."""
 
+import threading
+
 import six
 
+from google.cloud import exceptions
 from google.cloud.exceptions import NotFound
 from google.cloud._helpers import _datetime_from_microseconds
 from google.cloud.bigquery.dataset import Dataset
@@ -27,6 +30,7 @@
 from google.cloud.bigquery._helpers import UDFResourcesProperty
 from google.cloud.bigquery._helpers import _EnumProperty
 from google.cloud.bigquery._helpers import _TypedProperty
+import google.cloud.future.base
 
 
 class Compression(_EnumProperty):
@@ -82,7 +86,7 @@ class WriteDisposition(_EnumProperty):
     ALLOWED = (WRITE_APPEND, WRITE_TRUNCATE, WRITE_EMPTY)
 
 
-class _BaseJob(object):
+class _BaseJob(google.cloud.future.base.PollingFuture):
     """Base class for jobs.
 
     :type client: :class:`google.cloud.bigquery.client.Client`
@@ -90,6 +94,7 @@ class _BaseJob(object):
                    for the dataset (which requires a project).
     """
     def __init__(self, client):
+        super(_BaseJob, self).__init__()
         self._client = client
         self._properties = {}
 
@@ -131,6 +136,8 @@ class _AsyncJob(_BaseJob):
     def __init__(self, name, client):
         super(_AsyncJob, self).__init__(client)
         self.name = name
+        self._result_set = False
+        self._completion_lock = threading.Lock()
 
     @property
     def job_type(self):
@@ -273,6 +280,9 @@ def _set_properties(self, api_response):
         self._properties.clear()
         self._properties.update(cleaned)
 
+        # For Future interface
+        self._set_future_result()
+
     @classmethod
     def _get_resource_config(cls, resource):
         """Helper for :meth:`from_api_repr`
@@ -345,7 +355,7 @@ def exists(self, client=None):
             return True
 
     def reload(self, client=None):
-        """API call:  refresh job properties via a GET request
+        """API call:  refresh job properties via a GET request.
 
         See
         https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/get
@@ -371,12 +381,56 @@ def cancel(self, client=None):
                       ``NoneType``
         :param client: the client to use.  If not passed, falls back to the
                        ``client`` stored on the current dataset.
+
+        :rtype: bool
+        :returns: Boolean indicating that the cancel request was sent.
         """
         client = self._require_client(client)
 
         api_response = client._connection.api_request(
             method='POST', path='%s/cancel' % (self.path,))
         self._set_properties(api_response['job'])
+        return True
+
+    # The following methods implement the PollingFuture interface. Note that
+    # the methods above are from the pre-Future interface and are left for
+    # compatibility. The only "overloaded" method is :meth:`cancel`, which
+    # satisfies both interfaces.
+
+    def _set_future_result(self):
+        """Set the result or exception from the job if it is complete."""
+        # This must be done in a lock to prevent the polling thread
+        # and main thread from both executing the completion logic
+        # at the same time.
+        with self._completion_lock:
+            # If the operation isn't complete or if the result has already been
+            # set, do not call set_result/set_exception again.
+            # Note: self._result_set is set to True in set_result and
+            # set_exception, in case those methods are invoked directly.
+            if not self.state != 'DONE' or self._result_set:
+                return
+
+            if self.error_result:
+                exception = exceptions.GoogleCloudError(
+                    self.error_result, errors=self.errors)
+                self.set_exception(exception)
+            else:
+                self.set_result(self)
+
+    def done(self):
+        # Do not refresh is the state is already done, as the job will not
+        # change once complete.
+        if self.state != 'DONE':
+            self.reload()
+        return self.state == 'DONE'
+
+    def result(self, timeout=None):
+        if self.state is None:
+            self.begin()
+        return super(self, _AsyncJob).result(timeout=timeout)
+
+    def cancelled(self):
+        return False
 
 
 class _LoadConfiguration(object):

From adec1f99e12302d6a21fb287f72bfcd7ca7d73de Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Wed, 19 Jul 2017 10:27:47 -0700
Subject: [PATCH 2/8] Make QueryJob return QueryResults from result()

---
 bigquery/google/cloud/bigquery/job.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 19a616c1dd3f..44d29d6fb3f2 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -1189,3 +1189,8 @@ def results(self):
         """
         from google.cloud.bigquery.query import QueryResults
         return QueryResults.from_query_job(self)
+
+    def result(self, timeout=None):
+        super(QueryJob, self).result(timeout=timeout)
+        # Return a QueryResults instance instead of returning the job.
+        return self.results()

From d105de091147f1f3f28434b9cbac00666756657b Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Wed, 19 Jul 2017 10:37:06 -0700
Subject: [PATCH 3/8] Fix supercall

---
 bigquery/google/cloud/bigquery/job.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 44d29d6fb3f2..67e30fd5a1ef 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -427,7 +427,7 @@ def done(self):
     def result(self, timeout=None):
         if self.state is None:
             self.begin()
-        return super(self, _AsyncJob).result(timeout=timeout)
+        return super(_AsyncJob, self).result(timeout=timeout)
 
     def cancelled(self):
         return False

From 45f54002042f520e701fbd1b263cde3932a7d7d1 Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Thu, 20 Jul 2017 13:59:20 -0700
Subject: [PATCH 4/8] Finish tests

---
 bigquery/google/cloud/bigquery/job.py | 95 ++++++++++++++++++++-------
 bigquery/tests/unit/test_job.py       | 68 ++++++++++++++++++-
 2 files changed, 136 insertions(+), 27 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 67e30fd5a1ef..8ae2e84a337a 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -15,6 +15,7 @@
 """Define API Jobs."""
 
 import threading
+import warnings
 
 import six
 
@@ -86,17 +87,23 @@ class WriteDisposition(_EnumProperty):
     ALLOWED = (WRITE_APPEND, WRITE_TRUNCATE, WRITE_EMPTY)
 
 
-class _BaseJob(google.cloud.future.base.PollingFuture):
-    """Base class for jobs.
+class _AsyncJob(google.cloud.future.base.PollingFuture):
+    """Base class for asynchronous jobs.
+
+    :type name: str
+    :param name: the name of the job
 
     :type client: :class:`google.cloud.bigquery.client.Client`
     :param client: A client which holds credentials and project configuration
                    for the dataset (which requires a project).
     """
-    def __init__(self, client):
-        super(_BaseJob, self).__init__()
+    def __init__(self, name, client):
+        super(_AsyncJob, self).__init__()
+        self.name = name
         self._client = client
         self._properties = {}
+        self._result_set = False
+        self._completion_lock = threading.Lock()
 
     @property
     def project(self):
@@ -122,23 +129,6 @@ def _require_client(self, client):
             client = self._client
         return client
 
-
-class _AsyncJob(_BaseJob):
-    """Base class for asynchronous jobs.
-
-    :type name: str
-    :param name: the name of the job
-
-    :type client: :class:`google.cloud.bigquery.client.Client`
-    :param client: A client which holds credentials and project configuration
-                   for the dataset (which requires a project).
-    """
-    def __init__(self, name, client):
-        super(_AsyncJob, self).__init__(client)
-        self.name = name
-        self._result_set = False
-        self._completion_lock = threading.Lock()
-
     @property
     def job_type(self):
         """Type of job
@@ -407,10 +397,10 @@ def _set_future_result(self):
             # set, do not call set_result/set_exception again.
             # Note: self._result_set is set to True in set_result and
             # set_exception, in case those methods are invoked directly.
-            if not self.state != 'DONE' or self._result_set:
+            if self.state != 'DONE' or self._result_set:
                 return
 
-            if self.error_result:
+            if self.error_result is not None:
                 exception = exceptions.GoogleCloudError(
                     self.error_result, errors=self.errors)
                 self.set_exception(exception)
@@ -418,6 +408,11 @@ def _set_future_result(self):
                 self.set_result(self)
 
     def done(self):
+        """Refresh the job and checks if it is complete.
+
+        :rtype: bool
+        :returns: True if the job is complete, False otherwise.
+        """
         # Do not refresh is the state is already done, as the job will not
         # change once complete.
         if self.state != 'DONE':
@@ -425,11 +420,33 @@ def done(self):
         return self.state == 'DONE'
 
     def result(self, timeout=None):
+        """Start the job and wait for it to complete and get the result.
+
+        :type timeout: int
+        :param timeout: How long to wait for job to complete before raising
+            a :class:`TimeoutError`.
+
+        :rtype: _AsyncJob
+        :returns: This instance.
+
+        :raises: :class:`~google.cloud.exceptions.GoogleCloudError` if the job
+            failed or  :class:`TimeoutError` if the job did not complete in the
+            given timeout.
+        """
         if self.state is None:
             self.begin()
         return super(_AsyncJob, self).result(timeout=timeout)
 
     def cancelled(self):
+        """Check if the job has been cancelled.
+
+        This always returns False. It's not possible to check if a job was
+        cancelled in the API. This method is here to satisfy the interface
+        for :class:`google.cloud.future.Future`.
+
+        :rtype: bool
+        :returns: False
+        """
         return False
 
 
@@ -1181,7 +1198,7 @@ def from_api_repr(cls, resource, client):
         job._set_properties(resource)
         return job
 
-    def results(self):
+    def query_results(self):
         """Construct a QueryResults instance, bound to this job.
 
         :rtype: :class:`~google.cloud.bigquery.query.QueryResults`
@@ -1190,7 +1207,35 @@ def results(self):
         from google.cloud.bigquery.query import QueryResults
         return QueryResults.from_query_job(self)
 
+    def results(self):
+        """DEPRECATED.
+
+        This method is deprecated. Use :meth:`query_results` or :meth:`result`.
+
+        Construct a QueryResults instance, bound to this job.
+
+        :rtype: :class:`~google.cloud.bigquery.query.QueryResults`
+        :returns: The query results.
+        """
+        warnings.warn(
+            'QueryJob.results() is deprecated. Please use query_results() or '
+            'result().', DeprecationWarning)
+        return self.query_results()
+
     def result(self, timeout=None):
+        """Start the job and wait for it to complete and get the result.
+
+        :type timeout: int
+        :param timeout: How long to wait for job to complete before raising
+            a :class:`TimeoutError`.
+
+        :rtype: :class:`~google.cloud.bigquery.query.QueryResults`
+        :returns: The query results.
+
+        :raises: :class:`~google.cloud.exceptions.GoogleCloudError` if the job
+            failed or  :class:`TimeoutError` if the job did not complete in the
+            given timeout.
+        """
         super(QueryJob, self).result(timeout=timeout)
         # Return a QueryResults instance instead of returning the job.
-        return self.results()
+        return self.query_results()
diff --git a/bigquery/tests/unit/test_job.py b/bigquery/tests/unit/test_job.py
index 57d96bf8ae15..42a054e6d8b9 100644
--- a/bigquery/tests/unit/test_job.py
+++ b/bigquery/tests/unit/test_job.py
@@ -12,6 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
+import warnings
+
 import unittest
 
 
@@ -1514,15 +1517,76 @@ def test_from_api_repr_w_properties(self):
         self.assertIs(dataset._client, client)
         self._verifyResourceProperties(dataset, RESOURCE)
 
-    def test_results(self):
+    def test_query_results(self):
         from google.cloud.bigquery.query import QueryResults
 
         client = _Client(self.PROJECT)
         job = self._make_one(self.JOB_NAME, self.QUERY, client)
-        results = job.results()
+        results = job.query_results()
         self.assertIsInstance(results, QueryResults)
         self.assertIs(results._job, job)
 
+    def test_results_is_deprecated(self):
+        client = _Client(self.PROJECT)
+        job = self._make_one(self.JOB_NAME, self.QUERY, client)
+
+        with warnings.catch_warnings(record=True) as warned:
+            warnings.simplefilter('always')
+            job.results()
+            self.assertEqual(len(warned), 1)
+            self.assertIn('deprecated', str(warned[0]))
+
+    def test_result(self):
+        from google.cloud.bigquery.query import QueryResults
+
+        client = _Client(self.PROJECT)
+        job = self._make_one(self.JOB_NAME, self.QUERY, client)
+        job._properties['status'] = {'state': 'DONE'}
+
+        result = job.result()
+
+        self.assertIsInstance(result, QueryResults)
+        self.assertIs(result._job, job)
+
+    def test_result_invokes_begins(self):
+        begun_resource = self._makeResource()
+        done_resource = copy.deepcopy(begun_resource)
+        done_resource['status'] = {'state': 'DONE'}
+        connection = _Connection(begun_resource, done_resource)
+        client = _Client(self.PROJECT, connection=connection)
+        job = self._make_one(self.JOB_NAME, self.QUERY, client)
+
+        job.result()
+
+        self.assertEqual(len(connection._requested), 2)
+        begin_request, reload_request = connection._requested
+        self.assertEqual(begin_request['method'], 'POST')
+        self.assertEqual(reload_request['method'], 'GET')
+
+    def test_result_error(self):
+        from google.cloud import exceptions
+
+        client = _Client(self.PROJECT)
+        job = self._make_one(self.JOB_NAME, self.QUERY, client)
+        error_result = {
+            'debugInfo': 'DEBUG',
+            'location': 'LOCATION',
+            'message': 'MESSAGE',
+            'reason': 'REASON'
+        }
+        job._properties['status'] = {
+            'errorResult': error_result,
+            'errors': [error_result],
+            'state': 'DONE'
+        }
+        job._set_future_result()
+
+        with self.assertRaises(exceptions.GoogleCloudError) as exc_info:
+            job.result()
+
+        self.assertIsInstance(exc_info.exception, exceptions.GoogleCloudError)
+        self.assertEqual(exc_info.exception.errors, [error_result])
+
     def test_begin_w_bound_client(self):
         PATH = '/projects/%s/jobs' % (self.PROJECT,)
         RESOURCE = self._makeResource()

From 5c0b96f73df53e569594b56bab720fd7d7938a77 Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Fri, 21 Jul 2017 10:43:17 -0700
Subject: [PATCH 5/8] Address review comments

---
 bigquery/google/cloud/bigquery/job.py | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 8ae2e84a337a..c8baaec4b782 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -33,6 +33,8 @@
 from google.cloud.bigquery._helpers import _TypedProperty
 import google.cloud.future.base
 
+_DONE_STATE = 'DONE'
+
 
 class Compression(_EnumProperty):
     """Pseudo-enum for ``compression`` properties."""
@@ -380,6 +382,8 @@ def cancel(self, client=None):
         api_response = client._connection.api_request(
             method='POST', path='%s/cancel' % (self.path,))
         self._set_properties(api_response['job'])
+        # The Future interface requires that we return True if the *attempt*
+        # to cancel was successful.
         return True
 
     # The following methods implement the PollingFuture interface. Note that
@@ -397,7 +401,7 @@ def _set_future_result(self):
             # set, do not call set_result/set_exception again.
             # Note: self._result_set is set to True in set_result and
             # set_exception, in case those methods are invoked directly.
-            if self.state != 'DONE' or self._result_set:
+            if self.state != _DONE_STATE or self._result_set:
                 return
 
             if self.error_result is not None:
@@ -415,9 +419,9 @@ def done(self):
         """
         # Do not refresh is the state is already done, as the job will not
         # change once complete.
-        if self.state != 'DONE':
+        if self.state != _DONE_STATE:
             self.reload()
-        return self.state == 'DONE'
+        return self.state == _DONE_STATE
 
     def result(self, timeout=None):
         """Start the job and wait for it to complete and get the result.

From 8ee3af266d7202ca9dc0361e38bd1c23954eea14 Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Fri, 21 Jul 2017 11:21:04 -0700
Subject: [PATCH 6/8] Map errors to exceptions

---
 bigquery/google/cloud/bigquery/job.py | 43 +++++++++++++++++++++++++--
 bigquery/tests/unit/test_job.py       | 37 +++++++++++++++++++++--
 2 files changed, 75 insertions(+), 5 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index c8baaec4b782..a802969afc39 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -14,10 +14,12 @@
 
 """Define API Jobs."""
 
+import collections
 import threading
 import warnings
 
 import six
+from six.moves import http_client
 
 from google.cloud import exceptions
 from google.cloud.exceptions import NotFound
@@ -36,6 +38,41 @@
 _DONE_STATE = 'DONE'
 
 
+_ERROR_REASON_TO_EXCEPTION = {
+    'accessDenied': http_client.FORBIDDEN,
+    'backendError': http_client.INTERNAL_SERVER_ERROR,
+    'billingNotEnabled': http_client.FORBIDDEN,
+    'billingTierLimitExceeded': http_client.BAD_REQUEST,
+    'blocked': http_client.FORBIDDEN,
+    'duplicate': http_client.CONFLICT,
+    'internalError': http_client.INTERNAL_SERVER_ERROR,
+    'invalid': http_client.BAD_REQUEST,
+    'invalidQuery': http_client.BAD_REQUEST,
+    'notFound': http_client.NOT_FOUND,
+    'notImplemented': http_client.NOT_IMPLEMENTED,
+    'quotaExceeded': http_client.FORBIDDEN,
+    'rateLimitExceeded': http_client.FORBIDDEN,
+    'resourceInUse': http_client.BAD_REQUEST,
+    'resourcesExceeded': http_client.BAD_REQUEST,
+    'responseTooLarge': http_client.FORBIDDEN,
+    'stopped': http_client.OK,
+    'tableUnavailable': http_client.BAD_REQUEST,
+}
+
+_FakeResponse = collections.namedtuple('_FakeResponse', ['status'])
+
+
+def _error_result_to_exception(error_result):
+    """"""
+    reason = error_result.get('reason')
+    status_code = _ERROR_REASON_TO_EXCEPTION.get(
+        reason, http_client.INTERNAL_SERVER_ERROR)
+    # make_exception expects an httplib2 response object.
+    fake_response = _FakeResponse(status=status_code)
+    return exceptions.make_exception(
+        fake_response, b'', error_info=error_result, use_json=False)
+
+
 class Compression(_EnumProperty):
     """Pseudo-enum for ``compression`` properties."""
     GZIP = 'GZIP'
@@ -405,8 +442,7 @@ def _set_future_result(self):
                 return
 
             if self.error_result is not None:
-                exception = exceptions.GoogleCloudError(
-                    self.error_result, errors=self.errors)
+                exception = _error_result_to_exception(self.error_result)
                 self.set_exception(exception)
             else:
                 self.set_result(self)
@@ -451,7 +487,8 @@ def cancelled(self):
         :rtype: bool
         :returns: False
         """
-        return False
+        return (self.error_result is not None
+                and self.error_result.get('reason') == 'stopped')
 
 
 class _LoadConfiguration(object):
diff --git a/bigquery/tests/unit/test_job.py b/bigquery/tests/unit/test_job.py
index 42a054e6d8b9..a8f45a5c5d9f 100644
--- a/bigquery/tests/unit/test_job.py
+++ b/bigquery/tests/unit/test_job.py
@@ -15,9 +15,30 @@
 import copy
 import warnings
 
+from six.moves import http_client
 import unittest
 
 
+class TestErrorResultToException(unittest.TestCase):
+    def _call_fut(self, *args, **kwargs):
+        from google.cloud.bigquery import job
+        return job._error_result_to_exception(*args, **kwargs)
+
+    def test_simple(self):
+        error_result = {
+            'reason': 'invalid',
+            'meta': 'data'
+        }
+        exception = self._call_fut(error_result)
+        self.assertEqual(exception.code, http_client.BAD_REQUEST)
+        self.assertIn("'meta': 'data'", exception.message)
+
+    def test_missing_reason(self):
+        error_result = {}
+        exception = self._call_fut(error_result)
+        self.assertEqual(exception.code, http_client.INTERNAL_SERVER_ERROR)
+
+
 class _Base(object):
     PROJECT = 'project'
     SOURCE1 = 'http://example.com/source1.csv'
@@ -1517,6 +1538,18 @@ def test_from_api_repr_w_properties(self):
         self.assertIs(dataset._client, client)
         self._verifyResourceProperties(dataset, RESOURCE)
 
+    def test_cancelled(self):
+        client = _Client(self.PROJECT)
+        job = self._make_one(self.JOB_NAME, self.QUERY, client)
+        job._properties['status'] = {
+            'state': 'DONE',
+            'errorResult': {
+                'reason': 'stopped'
+            }
+        }
+
+        self.assertTrue(job.cancelled())
+
     def test_query_results(self):
         from google.cloud.bigquery.query import QueryResults
 
@@ -1572,7 +1605,7 @@ def test_result_error(self):
             'debugInfo': 'DEBUG',
             'location': 'LOCATION',
             'message': 'MESSAGE',
-            'reason': 'REASON'
+            'reason': 'invalid'
         }
         job._properties['status'] = {
             'errorResult': error_result,
@@ -1585,7 +1618,7 @@ def test_result_error(self):
             job.result()
 
         self.assertIsInstance(exc_info.exception, exceptions.GoogleCloudError)
-        self.assertEqual(exc_info.exception.errors, [error_result])
+        self.assertEqual(exc_info.exception.code, http_client.BAD_REQUEST)
 
     def test_begin_w_bound_client(self):
         PATH = '/projects/%s/jobs' % (self.PROJECT,)

From f9a71ba89e12980d26e6a33d34e52c00b0c09c5e Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Fri, 21 Jul 2017 11:36:22 -0700
Subject: [PATCH 7/8] Address review comments

---
 bigquery/google/cloud/bigquery/job.py | 19 ++++++++++++++++---
 bigquery/tests/unit/test_job.py       |  2 +-
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index a802969afc39..d08c6673b3fa 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -36,7 +36,7 @@
 import google.cloud.future.base
 
 _DONE_STATE = 'DONE'
-
+_STOPPED_REASON = 'stopped'
 
 _ERROR_REASON_TO_EXCEPTION = {
     'accessDenied': http_client.FORBIDDEN,
@@ -63,7 +63,20 @@
 
 
 def _error_result_to_exception(error_result):
-    """"""
+    """Maps BigQuery error reasons to an exception.
+
+    The reasons and their matching HTTP status codes are documented on
+    the `troubleshooting errors`_ page.
+
+    .. _troubleshooting errors: https://cloud.google.com/bigquery\
+        /troubleshooting-errors
+
+    :type error_result: Mapping[str, str]
+    :param error_result: The error result from BigQuery.
+
+    :rtype google.cloud.exceptions.GoogleCloudError:
+    :returns: The mapped exception.
+    """
     reason = error_result.get('reason')
     status_code = _ERROR_REASON_TO_EXCEPTION.get(
         reason, http_client.INTERNAL_SERVER_ERROR)
@@ -488,7 +501,7 @@ def cancelled(self):
         :returns: False
         """
         return (self.error_result is not None
-                and self.error_result.get('reason') == 'stopped')
+                and self.error_result.get('reason') == _STOPPED_REASON)
 
 
 class _LoadConfiguration(object):
diff --git a/bigquery/tests/unit/test_job.py b/bigquery/tests/unit/test_job.py
index a8f45a5c5d9f..d21471b4b656 100644
--- a/bigquery/tests/unit/test_job.py
+++ b/bigquery/tests/unit/test_job.py
@@ -19,7 +19,7 @@
 import unittest
 
 
-class TestErrorResultToException(unittest.TestCase):
+class Test__error_result_to_exception(unittest.TestCase):
     def _call_fut(self, *args, **kwargs):
         from google.cloud.bigquery import job
         return job._error_result_to_exception(*args, **kwargs)

From 7667c236f790e1f122ad921bac154e9de3fded42 Mon Sep 17 00:00:00 2001
From: Jon Wayne Parrott <jonwayne@google.com>
Date: Fri, 21 Jul 2017 15:22:40 -0700
Subject: [PATCH 8/8] Fix exception formatting, add system test

---
 bigquery/google/cloud/bigquery/job.py |  5 ++++-
 bigquery/tests/system.py              | 10 ++++++++++
 bigquery/tests/unit/test_job.py       |  5 +++--
 3 files changed, 17 insertions(+), 3 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index d08c6673b3fa..35a423b755b9 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -83,7 +83,10 @@ def _error_result_to_exception(error_result):
     # make_exception expects an httplib2 response object.
     fake_response = _FakeResponse(status=status_code)
     return exceptions.make_exception(
-        fake_response, b'', error_info=error_result, use_json=False)
+        fake_response,
+        error_result.get('message', ''),
+        error_info=error_result,
+        use_json=False)
 
 
 class Compression(_EnumProperty):
diff --git a/bigquery/tests/system.py b/bigquery/tests/system.py
index 3391ec2bd2d8..1d3da3d2a83d 100644
--- a/bigquery/tests/system.py
+++ b/bigquery/tests/system.py
@@ -19,6 +19,7 @@
 import os
 import time
 import unittest
+import uuid
 
 from google.cloud import bigquery
 from google.cloud._helpers import UTC
@@ -1013,6 +1014,15 @@ def test_large_query_w_public_data(self):
         rows = list(iterator)
         self.assertEqual(len(rows), LIMIT)
 
+    def test_async_query_future(self):
+        query_job = Config.CLIENT.run_async_query(
+            str(uuid.uuid4()), 'SELECT 1')
+        query_job.use_legacy_sql = False
+
+        iterator = query_job.result().fetch_data()
+        rows = list(iterator)
+        self.assertEqual(rows, [(1,)])
+
     def test_insert_nested_nested(self):
         # See #2951
         SF = bigquery.SchemaField
diff --git a/bigquery/tests/unit/test_job.py b/bigquery/tests/unit/test_job.py
index d21471b4b656..8b9d079df148 100644
--- a/bigquery/tests/unit/test_job.py
+++ b/bigquery/tests/unit/test_job.py
@@ -27,11 +27,12 @@ def _call_fut(self, *args, **kwargs):
     def test_simple(self):
         error_result = {
             'reason': 'invalid',
-            'meta': 'data'
+            'message': 'bad request'
         }
         exception = self._call_fut(error_result)
         self.assertEqual(exception.code, http_client.BAD_REQUEST)
-        self.assertIn("'meta': 'data'", exception.message)
+        self.assertTrue(exception.message.startswith('bad request'))
+        self.assertIn("'reason': 'invalid'", exception.message)
 
     def test_missing_reason(self):
         error_result = {}