From 9ac05c354e9f4be9426a428bd06adee425e1ac88 Mon Sep 17 00:00:00 2001
From: Pavan Kumar <gopidesupavan@gmail.com>
Date: Wed, 9 Jul 2025 23:19:12 +0100
Subject: [PATCH 1/2] Fix mypy gcp

---
 .../src/airflow/providers/google/cloud/hooks/datafusion.py    | 2 +-
 .../airflow/providers/google/cloud/links/kubernetes_engine.py | 3 +++
 .../src/airflow/providers/google/cloud/sensors/datafusion.py  | 4 ++--
 3 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py b/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py
index c228b3fa7c0e9..c27f63eb8399a 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py
+++ b/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),
diff --git a/providers/google/src/airflow/providers/google/cloud/links/kubernetes_engine.py b/providers/google/src/airflow/providers/google/cloud/links/kubernetes_engine.py
index 2b94326cbba0a..095bb0474a9a2 100644
--- a/providers/google/src/airflow/providers/google/cloud/links/kubernetes_engine.py
+++ b/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,
diff --git a/providers/google/src/airflow/providers/google/cloud/sensors/datafusion.py b/providers/google/src/airflow/providers/google/cloud/sensors/datafusion.py
index ea52f9af750c1..2f06c1144b95a 100644
--- a/providers/google/src/airflow/providers/google/cloud/sensors/datafusion.py
+++ b/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

From 1fa2c6a7149e2fd293a1c83812f77e9da68bb8d1 Mon Sep 17 00:00:00 2001
From: Pavan Kumar <gopidesupavan@gmail.com>
Date: Thu, 10 Jul 2025 17:29:33 +0100
Subject: [PATCH 2/2] Make google provider compatible with mypy 1.16.1

---
 .../src/airflow/providers/google/cloud/hooks/bigquery.py    | 2 +-
 .../src/airflow/providers/google/cloud/hooks/cloud_sql.py   | 4 ++++
 .../airflow/providers/google/cloud/operators/bigquery.py    | 6 ++++--
 .../src/airflow/providers/google/cloud/triggers/bigquery.py | 6 +++---
 4 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py b/providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py
index a0f635c584ab4..7faa955dc70ee 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py
+++ b/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:
         """
diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py b/providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py
index 1d8ac9e91d7b6..72a526e8c3133 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py
+++ b/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(),
diff --git a/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py b/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
index b245c2d4bf5b5..30d708a3c6006 100644
--- a/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
+++ b/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:
diff --git a/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py b/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py
index 2cac99543a637..b0ae15935b7cb 100644
--- a/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py
+++ b/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...")
