feat: add execution_project configuration (#9)

Author: maztohir

task/bq2bq/executor/bumblebee/bigquery_service.py

@@ -105,7 +105,7 @@ def transform_load(self,
             query_job_config.destination = table_ref
 
         logger.info("transform load")
-        query_job = self.client.query(query=query, job_config=query_job_config, project=source_project_id)
+        query_job = self.client.query(query=query, job_config=query_job_config)
         logger.info("Job {} is initially in state {} of {} project".format(query_job.job_id, query_job.state,
                                                                            query_job.project))
 
@@ -157,7 +157,7 @@ def create_bigquery_service(task_config: TaskConfigFromEnv, labels, writer):
              'https://www.googleapis.com/auth/cloud-platform',
              'https://www.googleapis.com/auth/drive')
     credentials, _ = google.auth.default(scopes=SCOPE)
-    client = bigquery.Client(project=task_config.destination_project, credentials=credentials)
+    client = bigquery.Client(project=task_config.execution_project, credentials=credentials)
     bigquery_service = BigqueryService(client, labels, writer)
     return bigquery_service
 

task/bq2bq/executor/bumblebee/config.py

@@ -113,6 +113,7 @@ class TaskConfigFromEnv(TaskConfig):
 
     def __init__(self):
         self._destination_project = get_env_config("PROJECT", raise_if_empty=True)
+        self._execution_project = get_env_config("EXECUTION_PROJECT", default=self._destination_project)
         self._destination_dataset = get_env_config("DATASET", raise_if_empty=True)
         self._destination_table_name = get_env_config("TABLE", raise_if_empty=True)
         self._sql_type = get_env_config("SQL_TYPE", raise_if_empty=True)
@@ -126,6 +127,10 @@ def __init__(self):
     def destination_project(self) -> str:
         return self._destination_project
 
+    @property
+    def execution_project(self) -> str:
+        return self._execution_project
+
     @property
     def destination_dataset(self) -> str:
         return self._destination_dataset
@@ -313,6 +318,7 @@ def __init__(self, raw_properties):
 
         self._destination_table_name = _validate_not_empty(self._get_property("TABLE"))
         self._destination_project = _validate_not_empty(self._get_property("PROJECT"))
+        self._execution_project = _validate_not_empty(self._get_property_or_default("EXECUTION_PROJECT", self._destination_project))
         self._destination_dataset = _validate_not_empty(self._get_property("DATASET"))
 
         self._window_size = _validate_window_size(self._get_property("WINDOW_SIZE"))
@@ -338,6 +344,10 @@ def destination_dataset(self):
     def destination_project(self):
         return self._destination_project
 
+    @property
+    def execution_project(self):
+        return self._execution_project
+
     @property
     def destination_table_name(self):
         return self._destination_table_name