From caeb35d2541b9b62beb2c50d135df2355e6bc254 Mon Sep 17 00:00:00 2001
From: Shahar Epstein <60007259+shahar1@users.noreply.github.com>
Date: Wed, 22 Apr 2026 10:20:58 +0300
Subject: [PATCH 1/3] Fix Google Dataflow hook failing import when apache-beam
 not installed

DataflowHook imported BeamHook, BeamRunnerType, and beam_options_to_args
at module level, making the entire google provider fail to import when
apache-airflow-providers-apache-beam is not installed.

apache-beam is not a declared dependency of the google provider, so this
hard import was always incorrect. The import error surfaced during the
provider release process (#65614) when the provider yaml check script
imports all registered provider modules and beam was not present in the
environment.

Move the imports to lazy form inside the two methods that actually use
them: DataflowHook.__init__ and DataflowHook._build_gcloud_command.
---
 .../src/airflow/providers/google/cloud/hooks/dataflow.py     | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
index 46788acb3cda6..d8f4c8a1305f3 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
@@ -50,7 +50,6 @@
 from googleapiclient.discovery import Resource, build
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType, beam_options_to_args
 from airflow.providers.common.compat.sdk import AirflowException, timeout
 from airflow.providers.google.common.hooks.base_google import (
     PROVIDE_PROJECT_ID,
@@ -624,6 +623,8 @@ def __init__(
         expected_terminal_state: str | None = None,
         **kwargs,
     ) -> None:
+        from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType
+
         self.poll_sleep = poll_sleep
         self.drain_pipeline = drain_pipeline
         self.cancel_timeout = cancel_timeout
@@ -986,6 +987,8 @@ def launch_beam_yaml_job(
         return job_id
 
     def _build_gcloud_command(self, command: list[str], parameters: dict[str, str]) -> list[str]:
+        from airflow.providers.apache.beam.hooks.beam import beam_options_to_args
+
         _parameters = deepcopy(parameters)
         if self.impersonation_chain:
             if isinstance(self.impersonation_chain, str):

From fbf2272ddaa0120689e9ab4d3dbdca962504c03e Mon Sep 17 00:00:00 2001
From: Shahar Epstein <60007259+shahar1@users.noreply.github.com>
Date: Wed, 22 Apr 2026 10:46:05 +0300
Subject: [PATCH 2/3] Fix lazy import

---
 .../providers/google/cloud/hooks/dataflow.py      | 15 ++++++++++-----
 1 file changed, 10 insertions(+), 5 deletions(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
index d8f4c8a1305f3..b7fb341aeccb4 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
@@ -50,7 +50,16 @@
 from googleapiclient.discovery import Resource, build
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.common.compat.sdk import AirflowException, timeout
+from airflow.providers.common.compat.sdk import (
+    AirflowException,
+    AirflowOptionalProviderFeatureException,
+    timeout,
+)
+
+try:
+    from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType, beam_options_to_args
+except ImportError as e:
+    raise AirflowOptionalProviderFeatureException(e)
 from airflow.providers.google.common.hooks.base_google import (
     PROVIDE_PROJECT_ID,
     GoogleBaseAsyncHook,
@@ -623,8 +632,6 @@ def __init__(
         expected_terminal_state: str | None = None,
         **kwargs,
     ) -> None:
-        from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType
-
         self.poll_sleep = poll_sleep
         self.drain_pipeline = drain_pipeline
         self.cancel_timeout = cancel_timeout
@@ -987,8 +994,6 @@ def launch_beam_yaml_job(
         return job_id
 
     def _build_gcloud_command(self, command: list[str], parameters: dict[str, str]) -> list[str]:
-        from airflow.providers.apache.beam.hooks.beam import beam_options_to_args
-
         _parameters = deepcopy(parameters)
         if self.impersonation_chain:
             if isinstance(self.impersonation_chain, str):

From a39983b448bbef15a8081776c39fd4756801d9cd Mon Sep 17 00:00:00 2001
From: Shahar Epstein <60007259+shahar1@users.noreply.github.com>
Date: Wed, 22 Apr 2026 11:56:16 +0300
Subject: [PATCH 3/3] Explicit message

---
 .../src/airflow/providers/google/cloud/hooks/dataflow.py    | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
index b7fb341aeccb4..2fb217a1729d6 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
@@ -59,7 +59,11 @@
 try:
     from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType, beam_options_to_args
 except ImportError as e:
-    raise AirflowOptionalProviderFeatureException(e)
+    raise AirflowOptionalProviderFeatureException(
+        "Failed to import apache-airflow-providers-apache-beam. "
+        "To use the Dataflow service with Apache Beam pipelines, please install the apache-beam provider: "
+        "pip install apache-airflow-providers-apache-beam"
+    ) from e
 from airflow.providers.google.common.hooks.base_google import (
     PROVIDE_PROJECT_ID,
     GoogleBaseAsyncHook,