Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion airflow/sensors/python.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,4 +71,7 @@ def poke(self, context: Context) -> PokeReturnValue | bool:

self.log.info("Poking callable: %s", str(self.python_callable))
return_value = self.python_callable(*self.op_args, **self.op_kwargs)
return PokeReturnValue(bool(return_value))
if isinstance(return_value, PokeReturnValue):
return return_value
else:
return PokeReturnValue(bool(return_value))
11 changes: 11 additions & 0 deletions tests/sensors/test_python.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import pytest

from airflow.exceptions import AirflowSensorTimeout
from airflow.sensors.base import PokeReturnValue
from airflow.sensors.python import PythonSensor
from tests.operators.test_python import BasePythonTest

Expand All @@ -41,6 +42,16 @@ def test_python_sensor_raise(self):
with pytest.raises(ZeroDivisionError):
self.run_as_task(lambda: 1 / 0)

def test_python_sensor_xcom(self):
with self.dag:
task = self.opcls(
task_id=self.task_id,
python_callable=lambda: PokeReturnValue(True, "xcom"),
**self.default_kwargs(),
)
poke_result = task.poke({})
assert poke_result.xcom_value == "xcom"

def test_python_callable_arguments_are_templatized(self):
"""Test PythonSensor op_args are templatized"""
# Create a named tuple and ensure it is still preserved
Expand Down