kacpermuda commented on code in PR #60268:
URL: https://github.com/apache/airflow/pull/60268#discussion_r2698701448
##########
providers/common/compat/src/airflow/providers/common/compat/standard/operators.py:
##########
@@ -17,18 +17,74 @@
from __future__ import annotations
+from typing import TYPE_CHECKING
+
from airflow.providers.common.compat._compat_utils import create_module_getattr
+from airflow.providers.common.compat.version_compat import (
+ AIRFLOW_V_3_0_PLUS,
+ AIRFLOW_V_3_1_PLUS,
+ AIRFLOW_V_3_2_PLUS,
+)
_IMPORT_MAP: dict[str, str | tuple[str, ...]] = {
# Re-export from sdk (which handles Airflow 2.x/3.x fallbacks)
"BaseOperator": "airflow.providers.common.compat.sdk",
+ "BaseAsyncOperator": "airflow.providers.common.compat.sdk",
"get_current_context": "airflow.providers.common.compat.sdk",
+ "is_async_callable": "airflow.providers.common.compat.sdk",
# Standard provider items with direct fallbacks
"PythonOperator": ("airflow.providers.standard.operators.python",
"airflow.operators.python"),
"ShortCircuitOperator": ("airflow.providers.standard.operators.python",
"airflow.operators.python"),
"_SERIALIZERS": ("airflow.providers.standard.operators.python",
"airflow.operators.python"),
}
+if TYPE_CHECKING:
+ from airflow.sdk.bases.decorator import is_async_callable
+ from airflow.sdk.bases.operator import BaseAsyncOperator
+elif AIRFLOW_V_3_2_PLUS:
+ from airflow.sdk.bases.decorator import is_async_callable
+ from airflow.sdk.bases.operator import BaseAsyncOperator
+else:
+ if AIRFLOW_V_3_0_PLUS:
+ from airflow.sdk import BaseOperator
+ else:
+ from airflow.models import BaseOperator
+
+ def is_async_callable(func) -> bool:
+ """Detect if a callable is an async function."""
+ import inspect
+ from functools import partial
+
+ while isinstance(func, partial):
+ func = func.func
+ return inspect.iscoroutinefunction(func)
+
+ class BaseAsyncOperator(BaseOperator):
+ """Stub for Airflow < 3.2 that raises a clear error."""
+
+ @property
+ def is_async(self) -> bool:
+ return True
+
+ if not AIRFLOW_V_3_1_PLUS:
+
+ @property
+ def xcom_push(self) -> bool:
+ return self.do_xcom_push
+
+ @xcom_push.setter
+ def xcom_push(self, value: bool):
+ self.do_xcom_push = value
Review Comment:
Is this part correct @dabla ? It's breaking my AF2 dags, when installing
standard provider from main. Should it have another if/elif for AF2?
Example dag:
```
from datetime import datetime
from airflow.providers.common.compat.assets import Asset
from airflow.providers.standard.operators.python import PythonOperator
from airflow import DAG
def do_nothing():
pass
DAG_ID = "test_dag"
with DAG(
dag_id=DAG_ID,
start_date=datetime(2021, 1, 1),
schedule=None,
catchup=False,
default_args={"retries": 0},
) as dag:
task = PythonOperator(
task_id="task",
python_callable=do_nothing,
inlets=[Asset(uri="s3://bucket2/dir2/file2.txt"),
Asset(uri="s3://bucket2/dir2/file3.txt")],
)
```
Error:
```
[2026-01-16, 14:20:00 UTC] {__init__.py:77} DEBUG - Lineage called with
inlets: [Dataset(uri='s3://bucket2/dir2/file2.txt', extra=None),
Dataset(uri='s3://bucket2/dir2/file3.txt', extra=None)], outlets: []
[2026-01-16, 14:20:00 UTC] {taskinstance.py:3336} ERROR - Task failed with
exception
Traceback (most recent call last):
File
"/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line
282, in _run_raw_task
TaskInstance._execute_task_with_callbacks(
File
"/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line
3194, in _execute_task_with_callbacks
self.task.post_execute(context=context, result=result)
File
"/usr/local/lib/python3.12/site-packages/airflow/lineage/__init__.py", line 88,
in wrapper
self.xcom_push(context, key=PIPELINE_INLETS, value=inlets)
TypeError: 'bool' object is not callable
```
Full logs:
[dag_id=test_dag_run_id=manual__2026-01-16T14_19_59.788247+00_00_task_id=task_attempt=1.log](https://github.com/user-attachments/files/24673818/dag_id.test_dag_run_id.manual__2026-01-16T14_19_59.788247%2B00_00_task_id.task_attempt.1.log)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
