Refactor BigQuery hook methods to use python library (#8631)

* Refactor create_external_table

* Refactor patch_table and add update_table

* Refactor run_grant_dataset_view_access

* Refactor run_upsert_table

* Refactor insert_all

* Refactor delete_table

* Refactor list_rows

* Fix types

* Fix test

* Refactor poll_job_complete

* Refactor cancel_query

* Refactor run_with_configuration

* Refactor run_* methods

* Fix self.project_id issue

* Fixup run_table_upsert

Author: turbaszek

UPDATING.md

@@ -128,6 +128,9 @@ The previous option used a colon(`:`) to split the module from function. Now the
 The change aims to unify the format of all options that refer to objects in the `airflow.cfg` file.
 
 ### Changes in BigQueryHook
+In general all hook methods are decorated with `@GoogleBaseHook.fallback_to_default_project_id` thus
+parameters to hook can only be passed via keyword arguments.
+
 - `create_empty_table` method accepts now `table_resource` parameter. If provided all
 other parameters are ignored.
 - `create_empty_dataset` will now use values from `dataset_reference` instead of raising error
@@ -137,6 +140,8 @@ changed.
 - `update_dataset` requires now new `fields` argument (breaking change)
 - `delete_dataset` has new signature (dataset_id, project_id, ...)
 previous one was (project_id, dataset_id, ...) (breaking change)
+- `get_tabledata` returns list of rows instead of API response in dict format. This method is deprecated in
+ favor of `list_rows`. (breaking change)
 
 ### Added mypy plugin to preserve types of decorated functions
 

airflow/providers/google/cloud/example_dags/example_bigquery_operations.py

@@ -38,7 +38,7 @@
 PROJECT_ID = os.environ.get("GCP_PROJECT_ID", "example-project")
 BQ_LOCATION = "europe-north1"
 
-DATASET_NAME = os.environ.get("GCP_BIGQUERY_DATASET_NAME", "test_dataset")
+DATASET_NAME = os.environ.get("GCP_BIGQUERY_DATASET_NAME", "test_dataset_operations")
 LOCATION_DATASET_NAME = "{}_location".format(DATASET_NAME)
 DATA_SAMPLE_GCS_URL = os.environ.get(
     "GCP_BIGQUERY_DATA_GCS_URL",
@@ -71,7 +71,7 @@
     # [START howto_operator_bigquery_delete_table]
     delete_table = BigQueryDeleteTableOperator(
         task_id="delete_table",
-        deletion_dataset_table="{}.test_table".format(DATASET_NAME),
+        deletion_dataset_table=f"{PROJECT_ID}.{DATASET_NAME}.test_table",
     )
     # [END howto_operator_bigquery_delete_table]
 
@@ -89,7 +89,7 @@
 
     # [START howto_operator_bigquery_delete_view]
     delete_view = BigQueryDeleteTableOperator(
-        task_id="delete_view", deletion_dataset_table=f"{DATASET_NAME}.test_view"
+        task_id="delete_view", deletion_dataset_table=f"{PROJECT_ID}.{DATASET_NAME}.test_view"
     )
     # [END howto_operator_bigquery_delete_view]
 

airflow/providers/google/cloud/example_dags/example_bigquery_queries.py

@@ -115,7 +115,7 @@
         task_id="get_data",
         dataset_id=DATASET_NAME,
         table_id=TABLE_1,
-        max_results="10",
+        max_results=10,
         selected_fields="value,name",
     )
     # [END howto_operator_bigquery_get_data]