This is an automated email from the ASF dual-hosted git repository.
ferruzzi 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 c9e175d31fb fix celery version compact issue (#65752)
c9e175d31fb is described below
commit c9e175d31fb6a3db4414d453ce92ac494e8bece0
Author: Anish Giri <[email protected]>
AuthorDate: Fri Apr 24 11:36:30 2026 -0500
fix celery version compact issue (#65752)
---
.../airflow/providers/celery/executors/celery_executor_utils.py | 9 +++++----
providers/celery/src/airflow/providers/celery/version_compat.py | 3 ++-
2 files changed, 7 insertions(+), 5 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 f16b7c226fa..e7e8f9ce8aa 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
@@ -47,6 +47,7 @@ from airflow.providers.celery.version_compat import (
AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_1_9_PLUS,
AIRFLOW_V_3_2_PLUS,
+ AIRFLOW_V_3_3_PLUS,
)
from airflow.providers.common.compat.sdk import AirflowException,
AirflowTaskTimeout, Stats, conf, timeout
from airflow.utils.log.logging_mixin import LoggingMixin
@@ -204,8 +205,8 @@ def on_celery_worker_ready(*args, **kwargs):
# and deserialization for us.
@app.task(name="execute_workload")
def execute_workload(input: str) -> None:
- if not AIRFLOW_V_3_2_PLUS:
- return _execute_workload_pre_3_2(input)
+ if not AIRFLOW_V_3_3_PLUS:
+ return _execute_workload_pre_3_3(input)
from celery.exceptions import Ignore
from pydantic import TypeAdapter
@@ -235,8 +236,8 @@ def execute_workload(input: str) -> None:
raise
-def _execute_workload_pre_3_2(input: str) -> None:
- """Fallback for Airflow < 3.2 which lacks BaseExecutor.run_workload()."""
+def _execute_workload_pre_3_3(input: str) -> None:
+ """Fallback for Airflow < 3.3 which lacks BaseExecutor.run_workload() and
ExecutorWorkload."""
from celery.exceptions import Ignore
from pydantic import TypeAdapter
diff --git a/providers/celery/src/airflow/providers/celery/version_compat.py
b/providers/celery/src/airflow/providers/celery/version_compat.py
index 6d0c6107451..83dd3980f08 100644
--- a/providers/celery/src/airflow/providers/celery/version_compat.py
+++ b/providers/celery/src/airflow/providers/celery/version_compat.py
@@ -29,5 +29,6 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
AIRFLOW_V_3_1_9_PLUS = get_base_airflow_version_tuple() >= (3, 1, 9)
AIRFLOW_V_3_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 0)
+AIRFLOW_V_3_3_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
-__all__ = ["AIRFLOW_V_3_0_PLUS", "AIRFLOW_V_3_1_9_PLUS", "AIRFLOW_V_3_2_PLUS"]
+__all__ = ["AIRFLOW_V_3_0_PLUS", "AIRFLOW_V_3_1_9_PLUS", "AIRFLOW_V_3_2_PLUS",
"AIRFLOW_V_3_3_PLUS"]
