dstandish commented on a change in pull request #20530: URL: https://github.com/apache/airflow/pull/20530#discussion_r776574854
########## File path: airflow/decorators/sensor.py ########## @@ -0,0 +1,93 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from inspect import signature +from typing import Any, Callable, Dict, Optional, Tuple + +from airflow.decorators.base import get_unique_task_id, task_decorator_factory +from airflow.sensors.base import BaseSensorOperator + + +class DecoratedSensorOperator(BaseSensorOperator): + """ + Wraps a Python callable and captures args/kwargs when called for execution. + + :param python_callable: A reference to an object that is callable + :type python_callable: python callable + :param task_id: task Id + :type task_id: str + :param op_kwargs: a dictionary of keyword arguments that will get unpacked + in your function (templated) + :type op_kwargs: dict + :param op_args: a list of positional arguments that will get unpacked when + calling your callable (templated) + :type op_args: list + :param kwargs_to_upstream: For certain operators, we might need to upstream certain arguments + that would otherwise be absorbed by the DecoratedOperator (for example python_callable for the + PythonOperator). This gives a user the option to upstream kwargs as needed. + :type kwargs_to_upstream: dict + """ + + template_fields = ('op_args', 'op_kwargs') + template_fields_renderers = {"op_args": "py", "op_kwargs": "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 = ('python_callable',) + + def __init__( + self, + *, + python_callable: Callable, + task_id: str, + op_args: Tuple[Any], + op_kwargs: Dict[str, Any], + **kwargs, + ) -> None: + kwargs.pop('multiple_outputs') + kwargs['task_id'] = get_unique_task_id(task_id, kwargs.get('dag'), kwargs.get('task_group')) + self.python_callable = python_callable + # Check that arguments can be binded + signature(python_callable).bind(*op_args, **op_kwargs) + self.op_args = op_args + self.op_kwargs = op_kwargs + super().__init__(**kwargs) + + def poke(self, context: Dict) -> bool: + return self.python_callable(*self.op_args, **self.op_kwargs) + + +def sensor(python_callable: Optional[Callable] = None, multiple_outputs: Optional[bool] = None, **kwargs): + """ + Wraps a function into an Airflow operator. + + Accepts kwargs for operator kwarg. Can be reused in a single DAG. + + :param python_callable: Function to decorate + :type python_callable: Optional[Callable] + :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. Review comment: 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 -- 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]
