use pipeline options to create bq wrapper for export reads

ahmedabu98 · ahmedabu98 · commit 24b59ebdaf57 · 2023-07-21T11:07:42.000-04:00
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py b/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py
@@ -150,7 +150,8 @@ def process(self, element):
 
     class CleanUpProjects(beam.DoFn):
       def process(self, unused_element, unused_signal, pipeline_details):
-        bq = bigquery_tools.BigQueryWrapper()
+        bq = bigquery_tools.BigQueryWrapper.from_pipeline_options(
+            input.pipeline.options)
         pipeline_details = pipeline_details[0]
         if 'temp_table_ref' in pipeline_details.keys():
           temp_table_ref = pipeline_details['temp_table_ref']
@@ -230,7 +231,8 @@ def _get_temp_dataset(self):
   def process(self,
               element: 'ReadFromBigQueryRequest') -> Iterable[BoundedSource]:
     bq = bigquery_tools.BigQueryWrapper(
-        temp_dataset_id=self._get_temp_dataset().datasetId)
+        temp_dataset_id=self._get_temp_dataset().datasetId,
+        client=bigquery_tools.BigQueryWrapper._bigquery_client(self.options))
 
     if element.query is not None:
       self._setup_temporary_dataset(bq, element)
