Add 'Client.get_job' API wrapper.

Author: tseaver

bigquery/google/cloud/bigquery/client.py

@@ -187,6 +187,35 @@ def job_from_resource(self, resource):
             return QueryJob.from_api_repr(resource, self)
         raise ValueError('Cannot parse job resource')
 
+    def get_job(self, job_name, project=None):
+        """Fetch a job for the project associated with this client.
+
+        See
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/get
+
+        :type job_name: str
+        :param job_name: Name of the job.
+
+        :type project: str
+        :param project:
+            project ID owning the job (defaults to the client's project)
+
+        :rtype: :class:`~google.cloud.bigquery.job._AsyncJob`
+        :returns:
+            Concrete job instance, based on the resource returned by the API.
+        """
+        extra_params = {'projection': 'full'}
+
+        if project is None:
+            project = self.project
+
+        path = '/projects/{}/jobs/{}'.format(project, job_name)
+
+        resource = self._connection.api_request(
+            method='GET', path=path, query_params=extra_params)
+
+        return self.job_from_resource(resource)
+
     def list_jobs(self, max_results=None, page_token=None, all_users=None,
                   state_filter=None):
         """List jobs for the project associated with this client.

bigquery/tests/unit/test_client.py

@@ -208,6 +208,70 @@ def test_job_from_resource_unknown_type(self):
         with self.assertRaises(ValueError):
             client.job_from_resource({'configuration': {'nonesuch': {}}})
 
+    def test_get_job_miss_w_explict_project(self):
+        from google.cloud.exceptions import NotFound
+
+        PROJECT = 'PROJECT'
+        OTHER_PROJECT = 'OTHER_PROJECT'
+        JOB_ID = 'NONESUCH'
+        creds = _make_credentials()
+        client = self._make_one(PROJECT, creds)
+        conn = client._connection = _Connection()
+
+        with self.assertRaises(NotFound):
+            client.get_job(JOB_ID, project=OTHER_PROJECT)
+
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/OTHER_PROJECT/jobs/NONESUCH')
+        self.assertEqual(req['query_params'], {'projection': 'full'})
+
+    def test_get_job_hit(self):
+        from google.cloud.bigquery.job import QueryJob
+
+        PROJECT = 'PROJECT'
+        JOB_ID = 'query_job'
+        DATASET = 'test_dataset'
+        QUERY_DESTINATION_TABLE = 'query_destination_table'
+        QUERY = 'SELECT * from test_dataset:test_table'
+        ASYNC_QUERY_DATA = {
+            'id': '{}:{}'.format(PROJECT, JOB_ID),
+            'jobReference': {
+                'projectId': PROJECT,
+                'jobId': 'query_job',
+            },
+            'state': 'DONE',
+            'configuration': {
+                'query': {
+                    'query': QUERY,
+                    'destinationTable': {
+                        'projectId': PROJECT,
+                        'datasetId': DATASET,
+                        'tableId': QUERY_DESTINATION_TABLE,
+                    },
+                    'createDisposition': 'CREATE_IF_NEEDED',
+                    'writeDisposition': 'WRITE_TRUNCATE',
+                }
+            },
+        }
+        creds = _make_credentials()
+        client = self._make_one(PROJECT, creds)
+        conn = client._connection = _Connection(ASYNC_QUERY_DATA)
+
+        job = client.get_job(JOB_ID)
+
+        self.assertIsInstance(job, QueryJob)
+        self.assertEqual(job.name, JOB_ID)
+        self.assertEqual(job.create_disposition, 'CREATE_IF_NEEDED')
+        self.assertEqual(job.write_disposition, 'WRITE_TRUNCATE')
+
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/projects/PROJECT/jobs/query_job')
+        self.assertEqual(req['query_params'], {'projection': 'full'})
+
     def test_list_jobs_defaults(self):
         import six
         from google.cloud.bigquery.job import LoadTableFromStorageJob
@@ -607,6 +671,11 @@ def __init__(self, *responses):
         self._requested = []
 
     def api_request(self, **kw):
+        from google.cloud.exceptions import NotFound
         self._requested.append(kw)
+
+        if len(self._responses) == 0:
+            raise NotFound("miss")
+
         response, self._responses = self._responses[0], self._responses[1:]
         return response