Add 'ExtractTableToStorageJob.from_api_repr' classmethod.

Preparation for creating jobs from resources in future 'client.list_jobs'.

gcloud/bigquery/job.py

@@ -825,6 +825,8 @@ class ExtractTableToStorageJob(_AsyncJob):
     :param client: A client which holds credentials and project configuration
                    for the dataset (which requires a project).
     """
+    _CONFIG_KEY = 'extract'
+
     def __init__(self, name, source, destination_uris, client):
         super(ExtractTableToStorageJob, self).__init__(name, client)
         self.source = source
@@ -877,17 +879,41 @@ def _build_resource(self):
                 'jobId': self.name,
             },
             'configuration': {
-                'extract': {
+                self._CONFIG_KEY: {
                     'sourceTable': source_ref,
                     'destinationUris': self.destination_uris,
                 },
             },
         }
-        configuration = resource['configuration']['extract']
+        configuration = resource['configuration'][self._CONFIG_KEY]
         self._populate_config_resource(configuration)
 
         return resource
 
+    @classmethod
+    def from_api_repr(cls, resource, client):
+        """Factory:  construct a job given its API representation
+
+        :type resource: dict
+        :param resource: dataset job representation returned from the API
+
+        :type client: :class:`gcloud.bigquery.client.Client`
+        :param client: Client which holds credentials and project
+                       configuration for the dataset.
+
+        :rtype: :class:`gcloud.bigquery.job.ExtractTableToStorageJob`
+        :returns: Job parsed from ``resource``.
+        """
+        name, config = cls._get_resource_config(resource)
+        source_config = config['sourceTable']
+        assert source_config['projectId'] == client.project
+        dataset = Dataset(source_config['datasetId'], client)
+        source = Table(source_config['tableId'], dataset)
+        destination_uris = config['destinationUris']
+        job = cls(name, source, destination_uris, client=client)
+        job._set_properties(resource)
+        return job
+
 
 class _AsyncQueryConfiguration(object):
     """User-settable configuration options for asynchronous query jobs.

gcloud/bigquery/test_job.py

@@ -1035,11 +1035,30 @@ def _getTargetClass(self):
         from gcloud.bigquery.job import ExtractTableToStorageJob
         return ExtractTableToStorageJob
 
+    def _makeResource(self, started=False, ended=False):
+        resource = super(TestExtractTableToStorageJob, self)._makeResource(
+            started, ended)
+        config = resource['configuration']['extract']
+        config['sourceTable'] = {
+            'projectId': self.PROJECT,
+            'datasetId': self.DS_NAME,
+            'tableId': self.SOURCE_TABLE,
+        }
+        config['destinationUris'] = [self.DESTINATION_URI]
+        return resource
+
     def _verifyResourceProperties(self, job, resource):
         self._verifyReadonlyResourceProperties(job, resource)
 
         config = resource.get('configuration', {}).get('extract')
 
+        self.assertEqual(job.destination_uris, config['destinationUris'])
+
+        table_ref = config['sourceTable']
+        self.assertEqual(job.source.project, table_ref['projectId'])
+        self.assertEqual(job.source.dataset_name, table_ref['datasetId'])
+        self.assertEqual(job.source.name, table_ref['tableId'])
+
         if 'compression' in config:
             self.assertEqual(job.compression,
                              config['compression'])
@@ -1084,6 +1103,61 @@ def test_ctor(self):
         self.assertTrue(job.field_delimiter is None)
         self.assertTrue(job.print_header is None)
 
+    def test_from_api_repr_missing_identity(self):
+        self._setUpConstants()
+        client = _Client(self.PROJECT)
+        RESOURCE = {}
+        klass = self._getTargetClass()
+        with self.assertRaises(KeyError):
+            klass.from_api_repr(RESOURCE, client=client)
+
+    def test_from_api_repr_missing_config(self):
+        self._setUpConstants()
+        client = _Client(self.PROJECT)
+        RESOURCE = {
+            'id': '%s:%s' % (self.PROJECT, self.DS_NAME),
+            'jobReference': {
+                'projectId': self.PROJECT,
+                'jobId': self.JOB_NAME,
+            }
+        }
+        klass = self._getTargetClass()
+        with self.assertRaises(KeyError):
+            klass.from_api_repr(RESOURCE, client=client)
+
+    def test_from_api_repr_bare(self):
+        self._setUpConstants()
+        client = _Client(self.PROJECT)
+        RESOURCE = {
+            'id': self.JOB_ID,
+            'jobReference': {
+                'projectId': self.PROJECT,
+                'jobId': self.JOB_NAME,
+            },
+            'configuration': {
+                'extract': {
+                    'sourceTable': {
+                        'projectId': self.PROJECT,
+                        'datasetId': self.DS_NAME,
+                        'tableId': self.SOURCE_TABLE,
+                    },
+                    'destinationUris': [self.DESTINATION_URI],
+                }
+            },
+        }
+        klass = self._getTargetClass()
+        job = klass.from_api_repr(RESOURCE, client=client)
+        self.assertTrue(job._client is client)
+        self._verifyResourceProperties(job, RESOURCE)
+
+    def test_from_api_repr_w_properties(self):
+        client = _Client(self.PROJECT)
+        RESOURCE = self._makeResource()
+        klass = self._getTargetClass()
+        dataset = klass.from_api_repr(RESOURCE, client=client)
+        self.assertTrue(dataset._client is client)
+        self._verifyResourceProperties(dataset, RESOURCE)
+
     def test_begin_w_bound_client(self):
         PATH = 'projects/%s/jobs' % self.PROJECT
         RESOURCE = self._makeResource()