Add impersonation_scopes to BigQuery

[ying-w]

closes: #33400

Currently scopes for google libraries are taken from connection parameters, this PR creates an additional parameter impersonation_scopes that will be used in situations when impersonation_chain is being used.

The flow to get credentials for gcp providers goes something like this

1. A library like bigquery will call get_credentials()

   airflow/airflow/providers/google/cloud/hooks/bigquery.py

   Line 163 in 00cd44e

   credentials=self.get_credentials(),

2. This is provided by GoogleBaseHook

   airflow/airflow/providers/google/common/hooks/base_google.py

   Line 307 in 00cd44e

   credentials, _ = self.get_credentials_and_project_id()

3. This then makes a call to self.scopes()

   airflow/airflow/providers/google/common/hooks/base_google.py

   Line 290 in 00cd44e

   scopes=self.scopes,

4. scopes will then return scopes from connection extras or default (in _get_scopes())

   airflow/airflow/providers/google/common/hooks/base_google.py

   Line 414 in 00cd44e

   def scopes(self) -> Sequence[str]:

Possible alternative solutions

1. Override scopes() definition in custom hook - as detailed in above issue BigQuery with impersonation_chain does not accept custom scopes #33400 (comment)
2. Introduce a parameter called scopes - it would be confusing what takes precedence if there is scopes in parameter and connection
3. Create a class for impersonation https://google-auth.readthedocs.io/en/master/reference/google.auth.impersonated_credentials.html

I would need some help testing and bringing this over the finish line

Should be able to do something like

bq_hook = BigQueryHook(
    gcp_conn_id="my_con",
    impersonation_chain="[email protected]",
    use_legacy_sql=False,
    location="us",
    api_resource_configs={"query": {"useQueryCache": False, "priority": "BATCH"}},
    scopes = (
    "https://www.googleapis.com/auth/cloud-platform",
    "https://www.googleapis.com/auth/drive",
    ),
)
bq_hook.scopes  # check that it got set

[eladkal]

Can you add unit test to cover the change?

[dirrao]

Checks are failing. Can you fix them? Can you add the unit test for the change suggested by @eladkal ?

[ying-w]

i'll do another pass at it sometime this week

[ying-w]

I need some help figuring out how to write a test. I see that there is a cloud/operators/test_bigquery.py file with a impersonation_chain call but no impersonation_chain call in cloud/hooks/test_bigquery.py

The other example I looked at

airflow/tests/providers/google/cloud/operators/test_cloud_storage_transfer_service.py

Line 265 in 952d04a

google_impersonation_chain=IMPERSONATION_CHAIN,

but I'm not quite sure what that line is doing

[shahar1]

cloud/hooks/test_bigquery.py

I need some help figuring out how to write a test. I see that there is a cloud/operators/test_bigquery.py file with a impersonation_chain call but no impersonation_chain call in cloud/hooks/test_bigquery.py

The other example I looked at

airflow/tests/providers/google/cloud/operators/test_cloud_storage_transfer_service.py

Line 265 in 952d04a

google_impersonation_chain=IMPERSONATION_CHAIN,

but I'm not quite sure what that line is doing

You could mock the impersonation chain as done in other tests of GCP hooks (hooks/test_compute.py, for example; there are more).
As for the other example, it is called google_impersonation_chain to differentiate it from a similar parameter in the transfer operator's target/source cloud platform (which is not necessarily GCP).

[ying-w]

i think i fixed the test