Fix output property missing for airflow version < 2.4.0

python-sdk/src/astro/sql/operators/cleanup.py

@@ -4,10 +4,12 @@
 from datetime import timedelta
 from typing import Any
 
+import airflow
 from airflow.decorators.base import get_unique_task_id
 from airflow.exceptions import AirflowException
 from airflow.models.baseoperator import BaseOperator
 from airflow.models.dagrun import DagRun
+from packaging.version import Version
 
 try:
     # Airflow >= 2.3
@@ -232,9 +234,18 @@ def resolve_tables_from_tasks(  # noqa: C901
                     and isinstance(task, MappedOperator)
                     and issubclass(task.operator_class, OPERATOR_CLASSES_WITH_TABLE_OUTPUT)
                 ):
-                    for t in task.output.resolve(context):
-                        if isinstance(t, BaseTable):
-                            res.append(t)
+                    if Version(airflow.__version__) >= Version("2.4.0"):
+                        for t in task.output.resolve(context):
+                            if isinstance(t, BaseTable):
+                                res.append(t)
+                    else:
+                        from airflow.models.xcom_arg import XComArg
+
+                        task_output = XComArg(operator=task)
+                        for t in task_output.resolve(context):
+                            if isinstance(t, BaseTable):
+                                res.append(t)
+
             except AirflowException:  # pragma: no cover
                 self.log.info(
                     "xcom output for %s not found. Will not clean up this task",