bigqueryhook.table_exists always falls back to the default GCP connection_id.

[nathadfield]

Apache Airflow version: 1.10.*

Environment:

• Cloud provider or hardware configuration: GCP
• OS (e.g. from /etc/os-release):
  PRETTY_NAME="Debian GNU/Linux 9 (stretch)"
  NAME="Debian GNU/Linux"
  VERSION_ID="9"
  VERSION="9 (stretch)"
  VERSION_CODENAME=stretch
  ID=debian
• Install tools:
• Others:

What happened:
When running the BigQueryTableExistenceSensor, the task always fails with an error that suggests it always falling back to the default GCP/BigQuery connectiond id.

test = BigQueryTableExistenceSensor(
task_id='test',
bigquery_conn_id='my_gcp_connection',
project_id='my_project',
dataset_id='my_dataset',
table_id='my_table'
)

[2020-07-20 18:14:16,227] {taskinstance.py:1294} ERROR - Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see https://cloud.google.com/docs/authentication/getting-started
Traceback (most recent call last):
File "/opt/airflow/airflow/models/taskinstance.py", line 1053, in _run_raw_task
self._prepare_and_execute_task_with_callbacks(context, session, task)
File "/opt/airflow/airflow/models/taskinstance.py", line 1164, in _prepare_and_execute_task_with_callbacks
result = self._execute_task(context, task_copy)
File "/opt/airflow/airflow/models/taskinstance.py", line 1199, in _execute_task
result = task_copy.execute(context=context)
File "/opt/airflow/airflow/sensors/base_sensor_operator.py", line 119, in execute
while not self.poke(context):
File "/opt/airflow/airflow/providers/google/cloud/sensors/bigquery.py", line 77, in poke
table_id=self.table_id)
File "/opt/airflow/airflow/providers/google/common/hooks/base_google.py", line 357, in inner_wrapper
return func(self, *args, **kwargs)
File "/opt/airflow/airflow/providers/google/cloud/hooks/bigquery.py", line 214, in table_exists
self.get_client().get_table(table_reference)
File "/opt/airflow/airflow/providers/google/cloud/hooks/bigquery.py", line 123, in get_client
credentials=self._get_credentials()
File "/usr/local/lib/python3.6/site-packages/google/cloud/bigquery/client.py", line 179, in init
project=project, credentials=credentials, _http=_http
File "/usr/local/lib/python3.6/site-packages/google/cloud/client.py", line 226, in init
_ClientProjectMixin.init(self, project=project)
File "/usr/local/lib/python3.6/site-packages/google/cloud/client.py", line 178, in init
project = self._determine_default(project)
File "/usr/local/lib/python3.6/site-packages/google/cloud/client.py", line 193, in _determine_default
return _determine_default_project(project)
File "/usr/local/lib/python3.6/site-packages/google/cloud/_helpers.py", line 186, in _determine_default_project
_, project = google.auth.default()
File "/usr/local/lib/python3.6/site-packages/google/auth/_default.py", line 333, in default
raise exceptions.DefaultCredentialsError(_HELP_MESSAGE)
google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials. Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application. For more information, please see https://cloud.google.com/docs/authentication/getting-started

I think it is because the project_id is not specified when making the call to get_client although this is apparently optional.

https://github.com/apache/airflow/blob/master/airflow/providers/google/cloud/hooks/bigquery.py#L214

How to reproduce it:
Simply create a task based on this operator as defined above.

[turbaszek]

As you mentioned on slack, the project_id should be passed to get_client in table_exists method of the hook. That's visible from your stack trace:

File "/usr/local/lib/python3.6/site-packages/google/cloud/bigquery/client.py", line 179, in init

The project_id is optional because all new method use @GoogleBaseHook.fallback_to_default_project_id which as name suggest will try to resolve the default value.

@nathadfield would you like to fix it?

[nathadfield]

@turbaszek Thanks! Yes, I'll fix it.

Is there anything you'd recommend to add from a test perspective too?

[turbaszek]

I think we should make sure that the project is passed as expected to the underlying client. And I would suggest to run / extend the existing system tests

[nathadfield]

@turbaszek Thanks! Writing good tests is something that I'm still getting my head around. Are you able to give me any helpful pointers as to what you'd like to see here?

[turbaszek]

You should extend those tests:

airflow/tests/providers/google/cloud/hooks/test_bigquery.py

Lines 103 to 115 in c8c52e6

	@mock.patch("airflow.providers.google.cloud.hooks.bigquery.Client")
	def test_bigquery_table_exists_true(self, mock_client):
	result = self.hook.table_exists(project_id=PROJECT_ID, dataset_id=DATASET_ID, table_id=TABLE_ID)
	mock_client.return_value.get_table.assert_called_once_with(TABLE_REFERENCE)
	assert result is True
	@mock.patch("airflow.providers.google.cloud.hooks.bigquery.Client")
	def test_bigquery_table_exists_false(self, mock_client):
	mock_client.return_value.get_table.side_effect = NotFound("Dataset not found")
	result = self.hook.table_exists(project_id=PROJECT_ID, dataset_id=DATASET_ID, table_id=TABLE_ID)
	mock_client.return_value.get_table.assert_called_once_with(TABLE_REFERENCE)
	assert result is False

with assuring that the mock_client was called with proper arguments. In this case with project_id.

I think that we can change mocking to mock get_client instead of Client, that should make testing easier.