Make google provider compatible with mypy 1.16.1

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

@@ -2248,7 +2248,7 @@ def value_check(
         self,
         sql: str,
         pass_value: Any,
-        records: list[Any],
+        records: list[Any] | None = None,
         tolerance: float | None = None,
     ) -> None:
         """

providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py

@@ -1103,6 +1103,8 @@ def _generate_connection_uri(self) -> str:
         return connection_uri
 
     def _get_instance_socket_name(self) -> str:
+        if self.project_id is None:
+            raise ValueError("The project_id should not be none")
         return self.project_id + ":" + self.location + ":" + self.instance
 
     def _get_sqlproxy_instance_specification(self) -> str:
@@ -1135,6 +1137,8 @@ def get_sqlproxy_runner(self) -> CloudSqlProxyRunner:
             raise ValueError("Proxy runner can only be retrieved in case of use_proxy = True")
         if not self.sql_proxy_unique_path:
             raise ValueError("The sql_proxy_unique_path should be set")
+        if self.project_id is None:
+            raise ValueError("The project_id should not be None")
         return CloudSqlProxyRunner(
             path_prefix=self.sql_proxy_unique_path,
             instance_specification=self._get_sqlproxy_instance_specification(),

providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py

@@ -434,7 +434,7 @@ def get_pipeline_workflow(
         pipeline_id: str,
         pipeline_type: DataFusionPipelineType = DataFusionPipelineType.BATCH,
         namespace: str = "default",
-    ) -> Any:
+    ) -> dict:
         url = os.path.join(
             self._base_url(instance_url, namespace),
             quote(pipeline_name),

providers/google/src/airflow/providers/google/cloud/links/kubernetes_engine.py

@@ -57,6 +57,9 @@ def persist(cls, context: Context, **value):
         if isinstance(cluster, dict):
             cluster = Cluster.from_json(json.dumps(cluster))
 
+        if not cluster:
+            raise ValueError("Cluster must be provided for KubernetesEngineClusterLink.")
+
         super().persist(
             context=context,
             cluster_name=cluster.name,

providers/google/src/airflow/providers/google/cloud/operators/bigquery.py

@@ -116,8 +116,10 @@ def get_db_hook(self: BigQueryCheckOperator) -> _BigQueryHookWithFlexibleProject
             impersonation_chain=self.impersonation_chain,
             labels=self.labels,
         )
+
+        # mypy assuming project_id is read only, as project_id is a property in GoogleBaseHook.
         if self.project_id:
-            hook.project_id = self.project_id
+            hook.project_id = self.project_id  # type:ignore[misc]
         return hook
 
 
@@ -1156,7 +1158,7 @@ def execute(self, context: Context):
                     "BigQueryHook.list_rows() returns iterator when return_iterator is False (default)"
                 )
             self.log.info("Total extracted rows: %s", len(rows))
-
+            table_data: list[dict[str, Any]] | list[Any]
             if self.as_dict:
                 table_data = [dict(row) for row in rows]
             else:

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

@@ -115,7 +115,7 @@ def poke(self, context: Context) -> bool:
                 pipeline_id=self.pipeline_id,
                 namespace=self.namespace,
             )
-            pipeline_status = pipeline_workflow["status"]
+            pipeline_status = pipeline_workflow.get("status")
         except AirflowNotFoundException:
             message = "Specified Pipeline ID was not found."
             raise AirflowException(message)
@@ -132,4 +132,4 @@ def poke(self, context: Context) -> bool:
         self.log.debug(
             "Current status of the pipeline workflow for %s: %s.", self.pipeline_id, pipeline_status
         )
-        return pipeline_status in self.expected_statuses
+        return pipeline_status is not None and pipeline_status in self.expected_statuses

providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py

@@ -591,9 +591,9 @@ async def run(self) -> AsyncIterator[TriggerEvent]:  # type: ignore[override]
                 if response_from_hook["status"] == "success":
                     query_results = await hook.get_job_output(job_id=self.job_id, project_id=self.project_id)
                     records = hook.get_records(query_results)
-                    records = records.pop(0) if records else None
-                    hook.value_check(self.sql, self.pass_value, records, self.tolerance)
-                    yield TriggerEvent({"status": "success", "message": "Job completed", "records": records})
+                    _records = records.pop(0) if records else None
+                    hook.value_check(self.sql, self.pass_value, _records, self.tolerance)
+                    yield TriggerEvent({"status": "success", "message": "Job completed", "records": _records})
                     return
                 elif response_from_hook["status"] == "pending":
                     self.log.info("Query is still running...")