Tidied up default_job_config creation

Author: withnale

sqlalchemy_bigquery/base.py

@@ -1081,7 +1081,6 @@ def __init__(
         self,
         arraysize=5000,
         credentials_path=None,
-        project_id=None,
         billing_project_id=None,
         location=None,
         credentials_info=None,
@@ -1095,7 +1094,7 @@ def __init__(
         self.credentials_path = credentials_path
         self.credentials_info = credentials_info
         self.credentials_base64 = credentials_base64
-        self.project_id = project_id
+        self.project_id = None
         self.billing_project_id = billing_project_id
         self.location = location
         self.identifier_preparer = self.preparer(self)
@@ -1119,18 +1118,20 @@ def _build_formatted_table_id(table):
         """Build '<dataset_id>.<table_id>' string using given table."""
         return "{}.{}".format(table.reference.dataset_id, table.table_id)
 
-    @staticmethod
-    def _add_default_dataset_to_job_config(job_config: QueryJobConfig, project_id, dataset_id):
-        # If dataset_id is set, then we know the job_config isn't None
-        if project_id:
-            dataset = ConnectionProperty(key='dataset_project_id', value=project_id)
-            job_config.connection_properties = [ dataset ]
-        if dataset_id:
-            # If project_id is missing, use default project_id for the current environment
+
+    def create_job_config(self, provided_config: QueryJobConfig):
+        project_id = self.project_id
+        if self.dataset_id is None and project_id == self.billing_project_id:
+            return provided_config
+        job_config = provided_config or QueryJobConfig()
+        if project_id != self.billing_project_id:
+            job_config.connection_properties = [ConnectionProperty(key="dataset_project_id", value=project_id)]
+        if self.dataset_id:
             if not project_id:
                 _, project_id = auth.default()
+            job_config.default_dataset = "{}.{}".format(project_id, self.dataset_id)
+        return job_config
 
-            job_config.default_dataset = "{}.{}".format(project_id, dataset_id)
 
     def do_execute(self, cursor, statement, parameters, context=None):
         kwargs = {}
@@ -1140,13 +1141,13 @@ def do_execute(self, cursor, statement, parameters, context=None):
 
     def create_connect_args(self, url):
         (
-            project_id,
+            self.project_id,
             location,
             dataset_id,
             arraysize,
             credentials_path,
             credentials_base64,
-            default_query_job_config,
+            provided_job_config,
             list_tables_page_size,
             user_supplied_client,
         ) = parse_url(url)
@@ -1157,17 +1158,10 @@ def create_connect_args(self, url):
         self.credentials_path = credentials_path or self.credentials_path
         self.credentials_base64 = credentials_base64 or self.credentials_base64
         self.dataset_id = dataset_id
-        from google.cloud.bigquery import QueryJobConfig
-        if default_query_job_config is None:
-            default_query_job_config = QueryJobConfig()
         if self.billing_project_id is None:
-            self.billing_project_id = project_id
-        if self.project_id is None:
-            self.project_id = project_id
+            self.billing_project_id = self.project_id
 
-        self._add_default_dataset_to_job_config(
-            default_query_job_config, project_id, dataset_id
-        )
+        default_query_job_config = self.create_job_config(provided_job_config)
 
         if user_supplied_client:
             # The user is expected to supply a client with