Add sensor decorator #20530

mingshi-wang · 2021-12-27T23:45:53Z

Added the @task.sensor decorator to convert a Python function to an instance of the BaseSensorOperator. Example usage of the decorator is:

@task.sensor(poke_interval=60, timeout=3600, mode="poke")
def f():
    # implement the condition 
    condition_met = ...
    return condition_met

closes: #20323

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

boring-cyborg · 2021-12-27T23:45:55Z

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
Be sure to read the Airflow Coding style.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: [email protected]
Slack: https://s.apache.org/airflow-slack

mik-laj · 2021-12-28T00:02:14Z

Can you also add docs?

mingshi-wang · 2021-12-28T00:17:03Z

Can you also add docs?

Sure! Added a section to the taskflow tutorial doc.

eladkal · 2021-12-28T07:15:24Z

Added the @task.sensor decorator to convert a Python function to an instance of the BaseSensorOperator

So this is really a decorator for PythonSensor isn't it?

airflow/airflow/sensors/python.py

Line 25 in b7086f9

class PythonSensor(BaseSensorOperator):

mingshi-wang · 2021-12-28T07:27:14Z

Added the @task.sensor decorator to convert a Python function to an instance of the BaseSensorOperator

So this is really a decorator for PythonSensor isn't it?

airflow/airflow/sensors/python.py

Line 25 in b7086f9

class PythonSensor(BaseSensorOperator):

Yes.

josh-fell · 2021-12-28T15:13:05Z

airflow/decorators/sensor.py

+    :param multiple_outputs: if set, function return value will be
+        unrolled to multiple XCom values. List/Tuples will unroll to xcom values
+        with index as key. Dict will unroll to xcom values with keys as XCom keys.
+        Defaults to False.


Suggested change

:param multiple_outputs: if set, function return value will be

unrolled to multiple XCom values. List/Tuples will unroll to xcom values

with index as key. Dict will unroll to xcom values with keys as XCom keys.

Defaults to False.

:param multiple_outputs: If set to True, the decorated function's return value will be unrolled to

multiple XCom values. Dict will unroll to XCom values with its keys as XCom keys.

Defaults to False.

Tuples and Lists are not currently supported for multiple_outputs.

Thanks! I removed "multiple_outputs" from the key word arg list since sensor operator doesn't have outputs.

@josh-fell re

Tuples and Lists are not currently supported for multiple_outputs.

are you sure? how come it indicates it is supported in the docstring e.g. for _PythonDecoratedOperator?

Separately....

It seems to me that it would be nice if we could optionally return an XCom value with python sensor

so the return type would be.... and this is a mouthful but... Optional[Union[bool, Tuple[bool, Any]]]

or something like that

so, if you return bool it's as normal

and when sensing is successful, you can return True, {"some": "information"}, i.e. a 2-tuple where the first element is the poke boolean and the second element is the xcom return.

update: draft PR for allowing return of xcom from poke method. with this in place we could allow return of XCom (even with multiple outputs) with this decorator.

i'll try to confirm this appoach is OK tomorrow and work to get it merged.

@potiuk is right - we cannot have poke() return PokeReturnValue without breaking backward compatibility. The alternative I can think of is to add a new type of Sensor and let the poke() method return "PokeReturnValue". I'd like to see better suggestions.

while i'd prefer to avoid a special class, i just want to point out that using something like PokeReturnValue would not be backward incompatible.

all sensors that do not return PokeReturnValue would continue to work just as they do. the issue Jarek highlights is, i think, a different one. namely, that a sensor which did make use of PokeReturnValue would not be usable on older versions of airflow. which makes sense cus it's a new feature.

but the concern i think is then, to release that sensor, do you also need to bump the min airflow version of the provider. maybe there would be a way around that. Namely, maybe it would be ok to include that (new) sensor in the provider, not bump min airflow, and just let only that sensor not be usable with older airflow versions -- perhaps raising a hellpful error message in that scenario.

separately, since we're only talking about new sensors (since old ones would continue to work exactly as they do), it would also be possible to design it in a backward compatible way. e.g. check airrflow version and use the appropriate return type. However, i suspect very strongly that, with such a sensor, the xcom behavior would be integral to its operation, so there would be no point in enabling this kind of compatibility with old airflow versions (since they would not support it anyway!). all of which leads me to conclude that this is maybe a non-issue.

in any case, there was some discussion of the matter, and opinion seems to be leaning against adding xcom support in sensor poke method. i plan to catchup on the latest discussions tomorrow. though support for your PR is good! just trying to sort out quickly whether we ought also to take this opportunity to support xcom too.

thanks

Yeah. Also I think there is a possibility that we might be more relaxed about "airlfow 2.1/2.2 compatibility". We seem to have more and more accumulated changes in Airflow tht might make Airflow 2.3 another release that we can set all our providers as min-version.

There are some very good reasons why we might want to do it for all providers for Airflow 2.3 and this opens up the really interesting possibility of making contract changes (and PokeReturnValue in this case might be back in game).

I started discussion here in the devlist: https://lists.apache.org/thread/csczm7xmnntdz9wtjbod8pqgt13zoggo

and if we decide that one of the next waves of community providers should be Airflow 2.3+ only - we might actually consider some more "bold" change here.

@dstandish @potiuk Thanks for the clarification! I look forward to the discussion. I'd be interested in working on supporting xcom with the sensor's poke API if we decide to go this route.

HI @dstandish, @potiuk - I have raised a PR #20656 to capture your idea of supporting XCom return values for the base sensor operator. Please have a look if you have time. Thanks!

dstandish · 2021-12-29T00:11:54Z

airflow/decorators/__init__.py

+        if name == "sensor":
+            return sensor


@dimberman can you comment on whether this is a good approach or whether instead we should do multiple inheritance with mixins as is done with @task.virtualenv? i lean towards supporting the approach taken here but interested if you want to advocate for the mixin approach

Can provider implement sensor decorator? If yes in that case that one will be used. I'm wondering if we should not initialise ProviderManager with _taskflow_decorators that would include predefined builtin decorators (or add builtins in the same way)? In this way we may probably use only one mechanism of registering decorators and skip mixin approach.

Hi @turbaszek, I agree that it is ideal to have a single mechanism to register sensor decorators, but the one defined in this PR is for the base sensor operator which is not in the provider package.

@mingshi-wang if my provider package will define sensor decorator it will override the builtin one (will be returned in L34) - is this expected behaviour? I wouldn't say so as this imposes some security risk in my opinion (I think I'm using airflow sensor but under the hood something else is being executed).

@turbaszek You are correct and this is a valid concern to me. A solution I can think of is to raise an error in the ProviderManager if "sensor" is registered by any provider. This makes "sensor" a reserved name but I would like to hear you suggestion.

I think the same goes for other builtins (python for example). We either register them in ProviderManager upfront and raise an error when adding provider decorators or check here in init for possible name clash.

@dimberman @kaxil @potiuk happy to hear your opinion here 🤔

turbaszek · 2022-01-04T19:28:42Z

docs/apache-airflow/tutorial_taskflow_api.rst

+      return check_spark_job_done(job_id)
+
+
+  wait_for_job(start_spark_job())


This is really lovely 🚀

uranusjr · 2022-01-05T04:51:41Z

airflow/decorators/sensor.py

+    # since we won't mutate the arguments, we should just do the shallow copy
+    # there are some cases we can't deepcopy the objects (e.g protobuf).
+    shallow_copy_attrs: Tuple[str, ...] = ('python_callable',)


Why does this need to be a tuple specifically, not a more general protocol e.g. Sequence or Collection?

Should this attribute be mutable?

Probably does not need to sicne subclasses (does it even make sense to subclass this?) would generally just override the entire attribute.

uranusjr · 2022-01-06T04:19:59Z

airflow/decorators/__init__.py

@@ -19,6 +19,7 @@

 from airflow.decorators.python import PythonDecoratorMixin, python_task  # noqa
 from airflow.decorators.python_virtualenv import PythonVirtualenvDecoratorMixin
+from airflow.decorators.sensor import sensor


This module naming scheme screams gotcha to me. Probably better to either put sensor into the python module instead, or rename the module to sensors (plural).

I agree with @uranusjr , personally +1 for moving it to python for consistency

uranusjr · 2022-01-10T02:04:13Z

docs/apache-airflow/tutorial_taskflow_api.rst

+You can apply the @task.sensor decorator to convert a regular Python function to an instance of the BaseSensorOperator
+class. The Python function implements the poke logic and returns a Boolean value just as the poke() method in the
+BaseSensorOperator does.


Suggested change

You can apply the @task.sensor decorator to convert a regular Python function to an instance of the BaseSensorOperator

class. The Python function implements the poke logic and returns a Boolean value just as the poke() method in the

BaseSensorOperator does.

You can apply the ``@task.sensor`` decorator to convert a regular Python function to an instance of the BaseSensorOperator

class. The Python function implements the poke logic and returns a Boolean value just as the ``poke()`` method in the

BaseSensorOperator does.

uranusjr · 2022-01-10T02:05:52Z

tests/decorators/test_sensor.py

+            task.sensor(not_callable)
+
+    def test_basic_sensor_success(self, dag_maker):
+        @task.sensor()


Suggested change

@task.sensor()

@task.sensor

Assuming the parathenses-less form works (I think it should since task_decorator_factory provides that?) this should be used for consistency. (Although I personally dislike it.)

github-actions · 2022-02-27T00:22:45Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.

hu-dabao · 2022-10-31T20:49:32Z

May I know why this PR isn't approved and is there a plan to provide sensor as a decorator?

potiuk · 2022-10-31T20:52:45Z

Because there another PR open (look for it in the PRS)

hu-dabao · 2022-10-31T20:58:51Z

Thanks for the quick response. To whom it may concern, the newest PR of the same matter is here: #22562

mingshi-wang mentioned this pull request Dec 27, 2021

Add a @task.sensor TaskFlow decorator #20323

Closed

2 tasks

mik-laj requested a review from turbaszek December 28, 2021 00:01

dstandish self-requested a review December 28, 2021 00:10

dstandish changed the title ~~Add basic sensor decorator~~ Add sensor decorator Dec 28, 2021

mingshi-wang force-pushed the decorator branch from 912d824 to 62ae96c Compare December 28, 2021 04:27

mingshi-wang requested review from kaxil and potiuk as code owners December 28, 2021 04:27

josh-fell reviewed Dec 28, 2021

View reviewed changes

mingshi-wang force-pushed the decorator branch 3 times, most recently from 2208e4d to d68fce9 Compare December 28, 2021 23:49

dstandish reviewed Dec 29, 2021

View reviewed changes

This was referenced Dec 29, 2021

Allow sensors to return XCom values with PokeReturnValue #20547

Closed

Allow sensors to return XCom values using tuple #20546

Closed

turbaszek requested a review from dimberman December 30, 2021 14:06

mingshi-wang force-pushed the decorator branch 2 times, most recently from 9c9f417 to e252b9a Compare January 4, 2022 08:03

turbaszek reviewed Jan 4, 2022

View reviewed changes

turbaszek requested a review from dstandish January 4, 2022 19:30

uranusjr reviewed Jan 5, 2022

View reviewed changes

uranusjr reviewed Jan 6, 2022

View reviewed changes

mingshi-wang mentioned this pull request Jan 9, 2022

Update base sensor operator to support XCOM return value #20656

Merged

Add basic sensor decorator

df806ca

mingshi-wang force-pushed the decorator branch from e252b9a to df806ca Compare January 9, 2022 21:39

uranusjr reviewed Jan 10, 2022

View reviewed changes

github-actions bot added the stale Stale PRs per the .github/workflows/stale.yml policy file label Feb 27, 2022

github-actions bot closed this Mar 7, 2022

		return check_spark_job_done(job_id)


		wait_for_job(start_spark_job())

Add sensor decorator #20530

Add sensor decorator #20530

Conversation

mingshi-wang commented Dec 27, 2021 • edited by dstandish Loading

boring-cyborg bot commented Dec 27, 2021

mik-laj commented Dec 28, 2021

mingshi-wang commented Dec 28, 2021 • edited Loading

eladkal commented Dec 28, 2021

mingshi-wang commented Dec 28, 2021

josh-fell Dec 28, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dstandish Dec 29, 2021 • edited by potiuk Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

potiuk Dec 30, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

turbaszek Jan 4, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Feb 27, 2022

hu-dabao commented Oct 31, 2022

potiuk commented Oct 31, 2022

hu-dabao commented Oct 31, 2022 • edited Loading

mingshi-wang commented Dec 27, 2021 •

edited by dstandish

Loading

mingshi-wang commented Dec 28, 2021 •

edited

Loading

josh-fell Dec 28, 2021 •

edited

Loading

dstandish Dec 29, 2021 •

edited by potiuk

Loading

potiuk Dec 30, 2021 •

edited

Loading

turbaszek Jan 4, 2022 •

edited

Loading

hu-dabao commented Oct 31, 2022 •

edited

Loading