Ensure that configuration properties are copied when (re)loading jobs.

Author: tseaver

bigquery/google/cloud/bigquery/_helpers.py

@@ -662,6 +662,18 @@ def to_api_repr(self):
         return resource
 
 
+def _query_param_from_api_repr(resource):
+    """Helper:  construct concrete query parameter from JSON resource."""
+    qp_type = resource['parameterType']
+    if 'arrayType' in qp_type:
+        klass = ArrayQueryParameter
+    elif 'structTypes' in qp_type:
+        klass = StructQueryParameter
+    else:
+        klass = ScalarQueryParameter
+    return klass.from_api_repr(resource)
+
+
 class QueryParametersProperty(object):
     """Custom property type, holding query parameter instances."""
 

bigquery/google/cloud/bigquery/job.py

@@ -28,9 +28,14 @@
 from google.cloud.bigquery.table import Table
 from google.cloud.bigquery.table import _build_schema_resource
 from google.cloud.bigquery.table import _parse_schema_resource
+from google.cloud.bigquery._helpers import ArrayQueryParameter
 from google.cloud.bigquery._helpers import QueryParametersProperty
+from google.cloud.bigquery._helpers import ScalarQueryParameter
+from google.cloud.bigquery._helpers import StructQueryParameter
+from google.cloud.bigquery._helpers import UDFResource
 from google.cloud.bigquery._helpers import UDFResourcesProperty
 from google.cloud.bigquery._helpers import _EnumProperty
+from google.cloud.bigquery._helpers import _query_param_from_api_repr
 from google.cloud.bigquery._helpers import _TypedProperty
 
 _DONE_STATE = 'DONE'
@@ -58,6 +63,22 @@
 }
 
 
+def _bool_or_none(value):
+    """Helper: deserialize boolean value from JSON string."""
+    if isinstance(value, bool):
+        return value
+    if value is not None:
+        return value.lower() in ['t', 'true', '1']
+
+
+def _int_or_none(value):
+    """Helper: deserialize int value from JSON string."""
+    if isinstance(value, int):
+        return value
+    if value is not None:
+        return int(value)
+
+
 def _error_result_to_exception(error_result):
     """Maps BigQuery error reasons to an exception.
 
@@ -303,6 +324,10 @@ def _scrub_local_properties(self, cleaned):
         """Helper:  handle subclass properties in cleaned."""
         pass
 
+    def _copy_configuration_properties(self, configuration):
+        """Helper:  assign subclass configuration properties in cleaned."""
+        raise NotImplementedError("Abstract")
+
     def _set_properties(self, api_response):
         """Update properties from resource in body of ``api_response``
 
@@ -322,6 +347,8 @@ def _set_properties(self, api_response):
 
         self._properties.clear()
         self._properties.update(cleaned)
+        configuration = cleaned['configuration'][self._JOB_TYPE]
+        self._copy_configuration_properties(configuration)
 
         # For Future interface
         self._set_future_result()
@@ -759,6 +786,28 @@ def _scrub_local_properties(self, cleaned):
         schema = cleaned.pop('schema', {'fields': ()})
         self.schema = _parse_schema_resource(schema)
 
+    def _copy_configuration_properties(self, configuration):
+        """Helper:  assign subclass configuration properties in cleaned."""
+        self.allow_jagged_rows = _bool_or_none(
+            configuration.get('allowJaggedRows'))
+        self.allow_quoted_newlines = _bool_or_none(
+            configuration.get('allowQuotedNewlines'))
+        self.autodetect = _bool_or_none(
+            configuration.get('autodetect'))
+        self.create_disposition = configuration.get('createDisposition')
+        self.encoding = configuration.get('encoding')
+        self.field_delimiter = configuration.get('fieldDelimiter')
+        self.ignore_unknown_values = _bool_or_none(
+            configuration.get('ignoreUnknownValues'))
+        self.max_bad_records = _int_or_none(
+            configuration.get('maxBadRecords'))
+        self.null_marker = configuration.get('nullMarker')
+        self.quote_character = configuration.get('quote')
+        self.skip_leading_rows = _int_or_none(
+            configuration.get('skipLeadingRows'))
+        self.source_format = configuration.get('sourceFormat')
+        self.write_disposition = configuration.get('writeDisposition')
+
     @classmethod
     def from_api_repr(cls, resource, client):
         """Factory:  construct a job given its API representation
@@ -869,6 +918,11 @@ def _build_resource(self):
 
         return resource
 
+    def _copy_configuration_properties(self, configuration):
+        """Helper:  assign subclass configuration properties in cleaned."""
+        self.create_disposition = configuration.get('createDisposition')
+        self.write_disposition = configuration.get('writeDisposition')
+
     @classmethod
     def from_api_repr(cls, resource, client):
         """Factory:  construct a job given its API representation
@@ -1002,6 +1056,14 @@ def _build_resource(self):
 
         return resource
 
+    def _copy_configuration_properties(self, configuration):
+        """Helper:  assign subclass configuration properties in cleaned."""
+        self.compression = configuration.get('compression')
+        self.destination_format = configuration.get('destinationFormat')
+        self.field_delimiter = configuration.get('fieldDelimiter')
+        self.print_header = _bool_or_none(
+            configuration.get('printHeader'))
+
     @classmethod
     def from_api_repr(cls, resource, client):
         """Factory:  construct a job given its API representation
@@ -1247,6 +1309,24 @@ def _scrub_local_properties(self, cleaned):
         configuration = cleaned['configuration']['query']
 
         self.query = configuration['query']
+
+    def _copy_configuration_properties(self, configuration):
+        """Helper:  assign subclass configuration properties in cleaned."""
+        self.allow_large_results = _bool_or_none(
+            configuration.get('allowLargeResults'))
+        self.flatten_results = _bool_or_none(
+            configuration.get('flattenResults'))
+        self.use_query_cache = _bool_or_none(
+            configuration.get('useQueryCache'))
+        self.use_legacy_sql = _bool_or_none(
+            configuration.get('useLegacySql'))
+
+        self.create_disposition = configuration.get('createDisposition')
+        self.priority = configuration.get('priority')
+        self.write_disposition = configuration.get('writeDisposition')
+        self.maximum_billing_tier = configuration.get('maximumBillingTier')
+        self.maximum_bytes_billed = configuration.get('maximumBytesBilled')
+
         dest_remote = configuration.get('destinationTable')
 
         if dest_remote is None:
@@ -1255,9 +1335,30 @@ def _scrub_local_properties(self, cleaned):
         else:
             dest_local = self._destination_table_resource()
             if dest_remote != dest_local:
-                dataset = self._client.dataset(dest_remote['datasetId'])
+                project = dest_remote['projectId']
+                dataset = self._client.dataset(
+                    dest_remote['datasetId'], project=project)
                 self.destination = dataset.table(dest_remote['tableId'])
 
+        def_ds = configuration.get('defaultDataset')
+        if def_ds is None:
+            if self.default_dataset is not None:
+                del self.default_dataset
+        else:
+            project = def_ds['projectId']
+            self.default_dataset = self._client.dataset(def_ds['datasetId'])
+
+        udf_resources = []
+        for udf_mapping in configuration.get(self._UDF_KEY, ()):
+            key_val, = udf_mapping.items()
+            udf_resources.append(UDFResource(key_val[0], key_val[1]))
+        self._udf_resources = udf_resources
+
+        self._query_parameters = [
+            _query_param_from_api_repr(mapping)
+            for mapping in configuration.get(self._QUERY_PARAMETERS_KEY, ())
+        ]
+
     @classmethod
     def from_api_repr(cls, resource, client):
         """Factory:  construct a job given its API representation

bigquery/tests/unit/test__helpers.py

@@ -1528,6 +1528,82 @@ def test_to_api_repr_w_nested_struct(self):
         self.assertEqual(param.to_api_repr(), EXPECTED)
 
 
+class Test__query_param_from_api_repr(unittest.TestCase):
+
+    @staticmethod
+    def _call_fut(resource):
+        from google.cloud.bigquery._helpers import _query_param_from_api_repr
+
+        return _query_param_from_api_repr(resource)
+
+    def test_w_scalar(self):
+        from google.cloud.bigquery._helpers import ScalarQueryParameter
+
+        RESOURCE = {
+            'name': 'foo',
+            'parameterType': {'type': 'INT64'},
+            'parameterValue': {'value': '123'},
+        }
+
+        parameter = self._call_fut(RESOURCE)
+
+        self.assertIsInstance(parameter, ScalarQueryParameter)
+        self.assertEqual(parameter.name, 'foo')
+        self.assertEqual(parameter.type_, 'INT64')
+        self.assertEqual(parameter.value, 123)
+
+    def test_w_array(self):
+        from google.cloud.bigquery._helpers import ArrayQueryParameter
+
+        RESOURCE = {
+            'name': 'foo',
+            'parameterType': {
+                'type': 'ARRAY',
+                'arrayType': {'type': 'INT64'},
+            },
+            'parameterValue': {
+                'arrayValues': [
+                    {'value': '123'},
+                ]},
+        }
+
+        parameter = self._call_fut(RESOURCE)
+
+        self.assertIsInstance(parameter, ArrayQueryParameter)
+        self.assertEqual(parameter.name, 'foo')
+        self.assertEqual(parameter.array_type, 'INT64')
+        self.assertEqual(parameter.values, [123])
+
+    def test_w_struct(self):
+        from google.cloud.bigquery._helpers import StructQueryParameter
+
+        RESOURCE = {
+            'name': 'foo',
+            'parameterType': {
+                'type': 'STRUCT',
+                'structTypes': [
+                    {'name': 'foo', 'type': {'type': 'STRING'}},
+                    {'name': 'bar', 'type': {'type': 'INT64'}},
+                ],
+            },
+            'parameterValue': {
+                'structValues': {
+                    'foo': {'value': 'Foo'},
+                    'bar': {'value': '123'},
+                }
+            },
+        }
+
+        parameter = self._call_fut(RESOURCE)
+
+        self.assertIsInstance(parameter, StructQueryParameter)
+        self.assertEqual(parameter.name, 'foo')
+        self.assertEqual(
+            parameter.struct_types, {'foo': 'STRING', 'bar': 'INT64'})
+        self.assertEqual(parameter.struct_values, {'foo': 'Foo', 'bar': 123})
+
+
+
 class Test_QueryParametersProperty(unittest.TestCase):
 
     @staticmethod