(airflow) branch main updated: bugfix: cannot import name 'workloads' for Airflow v2 (#47152)

[el...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 9a95da58e76 bugfix: cannot import name 'workloads' for Airflow v2 (#47152)
9a95da58e76 is described below

commit 9a95da58e766c9525d9de37e33d6a52ab50a448d
Author: Ash Berlin-Taylor <as...@apache.org>
AuthorDate: Thu Feb 27 15:42:16 2025 +0000

    bugfix: cannot import name 'workloads' for Airflow v2 (#47152)
    
    The intent behind this isinstance check was to deal with this, but it was
    missed that you can't import it on Airflow 3. Oops
    
    Closes #47139
---
 .../airflow/providers/celery/executors/celery_executor_utils.py    | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 93f80135639..dbc2d0cf64c 100644
--- a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++ b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -258,10 +258,11 @@ def send_task_to_executor(
     task_tuple: TaskInstanceInCelery,
 ) -> tuple[TaskInstanceKey, CommandType, AsyncResult | ExceptionWithTraceback]:
     """Send task to executor."""
-    from airflow.executors import workloads
-
     key, args, queue, task_to_run = task_tuple
-    if isinstance(args, workloads.BaseWorkload):
+
+    if AIRFLOW_V_3_0_PLUS:
+        if TYPE_CHECKING:
+            assert isinstance(args, workloads.BaseWorkload)
         args = (args.model_dump_json(),)
     try:
         with timeout(seconds=OPERATION_TIMEOUT):