Fix MyPY errors for google.cloud.example_dags (#20232)

Part of #19891

Author: potiuk

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

@@ -193,10 +193,7 @@
         task_id="update_table",
         dataset_id=DATASET_NAME,
         table_id="test_table",
-        fields=[
-            {"name": "emp_name", "type": "STRING", "mode": "REQUIRED"},
-            {"name": "salary", "type": "INTEGER", "mode": "NULLABLE"},
-        ],
+        fields=["emp_name", "salary"],
         table_resource={
             "friendlyName": "Updated Table",
             "description": "Updated Table",

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

@@ -31,6 +31,7 @@
 import os
 from datetime import datetime
 from pathlib import Path
+from typing import Any, Dict
 
 from future.backports.urllib.parse import urlparse
 
@@ -99,7 +100,7 @@
 # [END howto_operator_gcp_create_build_from_storage_body]
 
 # [START howto_operator_create_build_from_repo_body]
-create_build_from_repo_body = {
+create_build_from_repo_body: Dict[str, Any] = {
     "source": {"repo_source": {"repo_name": GCP_SOURCE_REPOSITORY_NAME, "branch_name": "main"}},
     "steps": [
         {

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

@@ -132,7 +132,9 @@
         task_id="failover-instance",
         location="europe-north1",
         instance=MEMORYSTORE_REDIS_INSTANCE_NAME_2,
-        data_protection_mode=FailoverInstanceRequest.DataProtectionMode.LIMITED_DATA_LOSS,
+        data_protection_mode=FailoverInstanceRequest.DataProtectionMode(
+            FailoverInstanceRequest.DataProtectionMode.LIMITED_DATA_LOSS
+        ),
         project_id=GCP_PROJECT_ID,
     )
     # [END howto_operator_failover_instance]

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

@@ -42,6 +42,7 @@
 
 import os
 from datetime import datetime
+from typing import Any, Dict
 
 from airflow import models
 from airflow.providers.google.cloud.operators.functions import (
@@ -72,7 +73,7 @@
 # [END howto_operator_gcf_deploy_body]
 
 # [START howto_operator_gcf_default_args]
-default_args = {'retries': 3}
+default_args: Dict[str, Any] = {'retries': 3}
 # [END howto_operator_gcf_default_args]
 
 # [START howto_operator_gcf_deploy_variants]

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

@@ -21,7 +21,7 @@
 """
 import os
 from datetime import datetime
-from typing import Dict
+from typing import Any, Dict
 
 from airflow import models
 from airflow.operators.bash import BashOperator
@@ -65,7 +65,7 @@
     tags=['example'],
     params={"model_name": MODEL_NAME},
 ) as dag:
-    hyperparams = {
+    hyperparams: Dict[str, Any] = {
         'goal': 'MAXIMIZE',
         'hyperparameterMetricTag': 'metric1',
         'maxTrials': 30,

airflow/providers/google/cloud/operators/dataproc_metastore.py

@@ -19,7 +19,7 @@
 """This module contains Google Dataproc Metastore operators."""
 
 from time import sleep
-from typing import Dict, Optional, Sequence, Tuple, Union
+from typing import Collection, Dict, Optional, Sequence, Tuple, Union
 
 from google.api_core.retry import Retry, exponential_sleep_generator
 from google.cloud.metastore_v1 import MetadataExport, MetadataManagementActivity
@@ -218,7 +218,7 @@ def __init__(
         project_id: str,
         region: str,
         service_id: str,
-        metadata_import: MetadataImport,
+        metadata_import: Union[MetadataImport, Dict[str, Collection[str]]],
         metadata_import_id: str,
         request_id: Optional[str] = None,
         retry: Optional[Retry] = None,

airflow/providers/google/cloud/operators/workflows.py

@@ -19,7 +19,7 @@
 import re
 import uuid
 from datetime import datetime, timedelta
-from typing import Dict, Optional, Sequence, Tuple, Union
+from typing import Dict, List, Optional, Sequence, Tuple, Union
 
 import pytz
 from google.api_core.exceptions import AlreadyExists
@@ -179,7 +179,7 @@ def __init__(
         workflow_id: str,
         location: str,
         project_id: Optional[str] = None,
-        update_mask: Optional[FieldMask] = None,
+        update_mask: Optional[Union[FieldMask, Dict[str, List[str]]]] = None,
         retry: Optional[Retry] = None,
         timeout: Optional[float] = None,
         metadata: Optional[Sequence[Tuple[str, str]]] = None,

airflow/providers/google/cloud/sensors/datafusion.py

@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 """This module contains a Google Cloud Data Fusion sensors."""
-from typing import Optional, Sequence, Set, Union
+from typing import Iterable, Optional, Sequence, Union
 
 from airflow.exceptions import AirflowException
 from airflow.providers.google.cloud.hooks.datafusion import DataFusionHook
@@ -69,10 +69,10 @@ def __init__(
         self,
         pipeline_name: str,
         pipeline_id: str,
-        expected_statuses: Set[str],
+        expected_statuses: Iterable[str],
         instance_name: str,
         location: str,
-        failure_statuses: Optional[Set[str]] = None,
+        failure_statuses: Optional[Iterable[str]] = None,
         project_id: Optional[str] = None,
         namespace: str = "default",
         gcp_conn_id: str = 'google_cloud_default',