[AIRFLOW-8057] [AIP-31] Add @task decorator #8962

casassg · 2020-05-21T21:45:29Z

Airflow AIP-31 task decorator implementation. This decorator should facilitate wrapping a function into an operator and use it as such in a DAG. Closes #8057. Closes #8056.

Should be used without args or with args/kwargs for the underlying operator:

@task
def simple_task(...):

@operator(dag=dag)
def simple_task(...):

Task ID should be the function name by default.
Decorator should return an instance of PythonFunctionalOperator. This can be used to set task dependencies. Ex:

@task 
def simple_task(...)
  pass

simple_task >> another_task

Adding documentation on @task and functional style dags in concepts.rst

Work done in collaboration from @turbaszek and @evgenyshulman

Make sure to mark the boxes below before creating PR: [x]

Description above provides context of the change
Unit tests coverage for changes (not needed for documentation changes)
Target Github ISSUE in description if exists
Commits follow "How to write a good git commit message"
Relevant documentation is updated including usage instructions.
I will engage committers as explained in Contribution Workflow Example.

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.
Read the Pull Request Guidelines for more information.

boring-cyborg · 2020-05-21T21:45:33Z

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/master/CONTRIBUTING.rst)
Here are some useful points:

Pay attention to the quality of your code (flake8, pylint 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.
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: dev@airflow.apache.org
Slack: https://apache-airflow-slack.herokuapp.com/

aoen

Wow what a first commit :). I think mine was a 1 line bug fix.

aoen · 2020-05-21T22:06:26Z

airflow/models/xcom_arg.py

@@ -83,7 +83,7 @@ def __getitem__(self, item):
        """
        Implements xcomresult['some_result_key']
        """
-        return XComArg(operator=self.operator, key=item)
+        return XComArg(operator=self.operator, key=str(item))


Why the cast? Seems being explicit about the type received here might be safer, and the cast can happen on the caller side, otherwise this weakens type safety of this arg for all XCOM use cases.

Key will always be a string as it's casted when pushing the field to the XCom DB. This is just to make things easier to use when doing multiple_outputs also to ensure consistency. This allows you to do:

res[1]

Which in reality is more transparent.

It was a change we missed in #8652

Any chance you could pull this (and the test for it) out to a separate PR please?

related discussion: databand-ai#5 (comment)

This small cast makes the multiple_output piece work more reliably and more transparently to the user. Not actually a new feature, but mostly a fix of what already got merged in XComArg.

I can break it into a separate PR but note that XCom class already does this when saving the key (not when retrieving it)

airflow/operators/python.py

aoen · 2020-05-21T22:43:15Z

airflow/operators/python.py

+    ) -> None:
+        # Check if we need to generate a new task_id
+        task_id = kwargs.get('task_id', None)
+        dag = kwargs.get('dag', None) or DagContext.get_current_dag()


Will this function play nicely if neither side of the or returns true and task_id is None, e.g. if someone initializes a task and then adds it to a DAG later?

I guess it doesn't really make sense with this pattern so specify dag_id later, so I think raising an exception if DAG is not specified would be reasonable (+accompanying unit test).

DAG can be not specified when initialized outside of DAG context. And you want to make it to bind when you call it. If you manually assign it the DAG, then it wont work thats true. Not sure if I should check this here.

aoen · 2020-05-21T22:44:21Z

airflow/operators/python.py

+    ) -> None:
+        # Check if we need to generate a new task_id
+        task_id = kwargs.get('task_id', None)
+        dag = kwargs.get('dag', None) or DagContext.get_current_dag()


Why default 'dag'/task_id to None?

dag -> because we may declare the task outside without an explicit dag (and we want to fallback to current_dag then.
task_id -> not really any reason. can switch it.

airflow/operators/python.py

aoen · 2020-05-21T23:19:23Z

airflow/operators/python.py

+        super().__init__(*args, **kwargs)
+        self.python_callable = python_callable
+        self.multiple_outputs = multiple_outputs
+        self._kwargs = kwargs


How does the use of kwargs relate to their deprecation in Airflow 2.0:
e.g. in the BaseOperator code:

if args or kwargs: # TODO remove *args and **kwargs in Airflow 2.0 warnings.warn( 'Invalid arguments were passed to {c} (task_id: {t}). ' 'Support for passing such arguments will be dropped in ' 'Airflow 2.0. Invalid arguments were:' '\n*args: {a}\n**kwargs: {k}'.format( c=self.__class__.__name__, a=args, k=kwargs, t=task_id), category=PendingDeprecationWarning, stacklevel=3 )

Does this need a deprecation warning too? Should we just not allow kwargs/args here in the first place, or is it needed for backwards compatibility?

This already calls BaseOperator.__init__ so guessing it will already give you a warning.

Should we just drop the kwargs/etc support in the new operator though since it's deprecated?

Yes, I think that would be better. +1, for dropping it.

Not sure what you mean here. How then is the operator supposed to pass down kwargs like owner that is a BaseOperator valid kwarg? If I remove kwargs here, then I won't be able to set owner in this operator.

aoen · 2020-05-21T23:24:12Z

airflow/operators/python.py

+
+    def __call__(self, *args, **kwargs):
+        # If args/kwargs are set, then operator has been called. Raise exception
+        if self._called:


Curious why the functions can't be reused seems a bit annoying for users, wonder if we can fix this (e.g. late-binding the task_ids or something...) or add a TODO, might be worth adding a comment here. ._checking if it was already called feels a bit hacky as it kind of couples the task execution with global state.

Mainly is bc of 1 function == 1 operator. The alternative was the idea that @evgenyshulman proposed which is to use functions as operators generators. This simplifies a bit, but also then you can't use the change the operator later on (it's never accessible in the DAG file itself). I've seen both approaches take here (either 1to1 or 1tomany). For me I like better the 1to1, but mainly bc it allows you to use it as you would use any operator later and feels a bit more intuitive.

Is this a exeuction time error, or does this happen at parse time?

We don't appear to have any tests that cover this (or I missed it). Can you add some, and add a section about this to the docs.

Do I correctly understand that this will not work?

@task def update_user(user_id: str): ... with DAG(...): # Fetch list of users ... # Execute task for each user for user_id in users_list: update_user(user_id)

It will fail at parse time.

Correct. 1 function == 1 operator. Airflow doesn't allow dynamic operators (execute 1 operator several times). You can still work around it.

This will work though:

@task def update_user(user_id: str): ... with DAG(...): # Fetch list of users ... # Execute task for each user for user_id in users_list: update_user.copy(f'update_{user_id}')(user_id)

Hm, I have mixed feelings:

copy is not self-explanatory in this case imho

in such case, shouldn't we generate auto id? Or at least can we try to do update_user(user_id, task_id=user_id)

Agree with turbaszek. This is not self-explanatory at all. I think having update_user(user_id, task_id=...) would be much better. We can access the function signature inside task. This should work and is more self-explanatory.

Main worry is that then what is update_user. What you are describing here is using update_user as an operator factory. It has it's value, but it also feels too magic to me atm. If update_user is a factory, then you can't change the operator instance at all or use it to set non-data dependencies.

We could capture task_id kwarg and generate a new operator, but then what is update_user the first operator, the latest one? What does update_user represent?

You can either do (1) update_user(i) for i in range(20) or (2) update_user >> other_operation, but not both. I prefer to support 2nd option as it adapts more to what Airflow already does with operators.

We could capture task_id kwarg and generate a new operator, but then what is update_user the first operator, the latest one? What does update_user represent?

For me update_user is a function and as a function it can be called many times with different input thus yielding different results (here creating new task). I have never meet "function as a singleton" pattern. If we don't want to generate task_id for users then we may consider raising an exception on second invocation when no custom task_id is passed.

My point is: this is a function, I expect to be able to call it as many time as I wish. I expect Airflow to treat each call of this function (in proper context) as creating a new task.

You can either do (1) update_user(i) for i in range(20) or (2) update_user >> other_operation, but not both. I prefer to support 2nd option as it adapts more to what Airflow already does with operators.

Why should I not be able to do this? This is something that I saw many times.

first_task = BashOperator() last_task = BashOperator() for user_id in users_list: first_task >> update_user(user_id) >> last_task

Yeah, this use case is exactly what we want to support.

The issue being we'd have to generate a new task for each one (doable, that isn't a problem, we just need to work out what task_id to call it).

Perhaps taking a leaf out of pytest or https://pypi.org/project/parameterized/ for how they name functions/test cases?

aoen · 2020-05-21T23:27:56Z

airflow/operators/python.py

+            raise AirflowException('@task decorated functions can only be called once. If you need to reuse '
+                                   'it several times in a DAG, use the `copy` method.')
+
+        # If we have no DAG, reinitialize class to capture DAGContext and DAG default args.


Not great that this and potential parse-time errors that could occur here is moved to runtime instead of DAG parse time, another reason it might be worth thinking about a more parse-time friendly solution if possible (or what's stopping Airflow from supporting this at the current time).

I'm open to other options. This seemed the cleanest one. We are basically calling init again so that we capture default_args from the dag. The other option is to manually implement default_args here.

Also note that we will get parse errors in the declaration if there's any. So this will be for default_args itself.

What use case/code path is this enabling?

Defining a task decorated operator without a DAG and adding it to the DAG on __call__.

@taks def add_2(num) return num+2 with DAG(...): add_2(2)

Otherwise this does not work. Also if we define default_args in DAG we wont be able to capture it either.

Oh gotcha, that is a good workflow to enable.

Is this covered in tests? (Sorry, finding it hard to keep track of all the discussions at the moment)

Same tbh. I think given the discussion on calling a decorated function several times, I'll refactor to accomodate that. This should simplify code a bit as well.

Also, not sure if we should use the same approach from parametrized, as you may want to run an operation twice with the same args/kwargs. Also, if the arg/kwarg is a XComArg, what should we do?

airflow/operators/python.py

turbaszek · 2020-05-22T09:48:20Z

airflow/operators/python.py

+
+    def __call__(self, *args, **kwargs):
+        # If args/kwargs are set, then operator has been called. Raise exception
+        if self._called:


Do I correctly understand that this will not work?

@task def update_user(user_id: str): ... with DAG(...): # Fetch list of users ... # Execute task for each user for user_id in users_list: update_user(user_id)

airflow/operators/python.py

airflow/ti_deps/deps/trigger_rule_dep.py

ashb

This will be ace!

airflow/operators/python.py

ashb · 2020-05-22T09:27:51Z

airflow/operators/python.py

+            raise AirflowException('@task decorated functions can only be called once. If you need to reuse '
+                                   'it several times in a DAG, use the `copy` method.')
+
+        # If we have no DAG, reinitialize class to capture DAGContext and DAG default args.


What use case/code path is this enabling?

airflow/operators/python.py

ashb · 2020-05-22T09:32:59Z

airflow/operators/python.py

+
+    def __call__(self, *args, **kwargs):
+        # If args/kwargs are set, then operator has been called. Raise exception
+        if self._called:


Is this a exeuction time error, or does this happen at parse time?

We don't appear to have any tests that cover this (or I missed it). Can you add some, and add a section about this to the docs.

airflow/operators/python.py

ashb · 2020-05-22T10:27:33Z

tests/operators/test_python.py

+            do_run()
+            assert ['do_run'] == self.dag.task_ids
+            do_run_1 = do_run.copy()
+            do_run_2 = do_run.copy()
+        assert do_run_1.task_id == 'do_run__1'
+        assert do_run_2.task_id == 'do_run__2'


Suggested change

do_run()

assert ['do_run'] == self.dag.task_ids

do_run_1 = do_run.copy()

do_run_2 = do_run.copy()

assert do_run_1.task_id == 'do_run__1'

assert do_run_2.task_id == 'do_run__2'

run = do_run()

do_run_1 = do_run.copy()

do_run_2 = do_run.copy()

assert ['do_run', 'do_run__1', 'do_run__2'] == self.dag.task_ids

assert run.task_id == 'do_run'

assert do_run_1.task_id == 'do_run__1'

assert do_run_2.task_id == 'do_run__2'

(I think)

mostly, except:

assert do_run.task_id == 'do_run'

run here is an XComArg representing the value returned in the do_run operator.

ashb · 2020-05-22T10:28:48Z

tests/operators/test_python.py

+        def return_dict(number: int):
+            return {
+                'number': number + 1,
+                43: 43


We should test (and work out what we want) when you do return { 43: 43, '43': 42 }

XCom automatically casts keys to strings. That's the reason for adding casting in XComArg/

Yes, I mean we should define, and test, what we want the behaviour to be in this case.

Ooh I see, sorry did not fully understand the issue. Not sure what the best path for this may be. We can raise an exception (not great) or log weird usage.

tests/operators/test_python.py

ashb · 2020-05-22T10:35:34Z

Do copy what I put in slack, Thanks to Stackoverflow I've got a way of having from airflow import task be callable, and still work as a submodule:

diff --git airflow/task/__init__.py airflow/task/__init__.py
index 114d189da..0b15c9a8c 100644
--- airflow/task/__init__.py
+++ airflow/task/__init__.py
@@ -16,3 +16,20 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+
+import sys
+import types
+
+
+class CallableModule(types.ModuleType):
+    def __init__(self):
+        types.ModuleType.__init__(self, __name__)
+        self.__dict__.update(sys.modules[__name__].__dict__)
+
+    __all__ = list(set(vars().keys()) - {'__qualname__'})   # for python 2 and 3
+
+    def __call__(self, *args, **kwargs):
+        from airlfow.decorators import task
+        return task(*args, **kwargs)
+
+sys.modules[__name__] = CallableModule()

This works in Py 2.7, 3.7 and 3.8 (versions I have easy access to).

The main question is: do we want to support that? Is this "hack" worth it, and do IDEs get massively confused by this?

ashb · 2020-05-22T11:35:34Z

The main question is: do we want to support that? Is this "hack" worth it, and do IDEs get massively confused by this?

I think if IDEs support it, and we add tests for it I'd say, yes.

I wonder if we can do this in airflow/__init__.py (adding the task decorator import)

STATICA_HACK = True
globals()['kcah_acitats'[::-1].upper()] = False
if STATICA_HACK:  # pragma: no cover
    from airflow.models.dag import DAG
    from airflow.exceptions import AirflowException
    from airflow.decorators import task

Without confusing IDEs more.

casassg · 2020-05-22T17:28:50Z

I prefer to avoid doing that. Seems to add quite a bit of code complexity and I doubt it will impact significantly the user experience. I think it should be fine from users to use airflow.decorators.task

airflow/operators/python.py

feluelle · 2020-05-27T11:05:16Z

airflow/operators/python.py

+        super().__init__(*args, **kwargs)
+        self.python_callable = python_callable
+        self.multiple_outputs = multiple_outputs
+        self._kwargs = kwargs


Yes, I think that would be better. +1, for dropping it.

feluelle · 2020-05-27T11:11:19Z

airflow/operators/python.py

+
+    def __call__(self, *args, **kwargs):
+        # If args/kwargs are set, then operator has been called. Raise exception
+        if self._called:


Agree with turbaszek. This is not self-explanatory at all. I think having update_user(user_id, task_id=...) would be much better. We can access the function signature inside task. This should work and is more self-explanatory.

airflow/operators/python.py

docs/concepts.rst

feluelle · 2020-05-28T08:33:38Z

airflow/operators/python.py

+    @staticmethod
+    def _get_unique_task_id(task_id: str, dag: Optional[DAG]) -> str:
+        dag = dag or DagContext.get_current_dag()
+        if not dag or task_id not in dag.task_ids:
+            return task_id
+        core = re.split(r'__\d+$', task_id)[0]
+        suffixes = sorted(
+            [int(re.split(r'^.+__', task_id)[1])
+             for task_id in dag.task_ids
+             if re.match(rf'^{core}__\d+$', task_id)]
+        )
+        if not suffixes:
+            return f'{core}__1'
+        return f'{core}__{suffixes[-1] + 1}'


You are using F-Strings here, too - not only in examples.

we also use type annotations which is py3 only. We should probs decide if we want to support Python2 or not at all.

We can use f-strings

Yeah, we can use f-strings here, and manually fix it up when backporting -- it's not too much work on the release-manager/who ever does the backport.

When it comes time to backport this I can run someone through the process (or better yet, I should document this process)

casassg · 2020-05-28T18:55:40Z

Switched to use factory method (aka reuse function several times in a DAG). Cleaned a lot of code that had to do with this decision.
Added check for multiple outputs keys. Now if any key is not a string, it will fail. This reduces weird cases where users may accidentally duplicate keys when casted to strings. Less flexible but also easier to see what's happenning below on the XCom table.
Reverted string casting in XComArg as we no longer need as we are enforcing keys to always be strings

airflow/operators/python.py

turbaszek · 2020-05-29T11:04:05Z

airflow/operators/python.py

+    @staticmethod
+    def _get_unique_task_id(task_id: str, dag: Optional[DAG]) -> str:
+        dag = dag or DagContext.get_current_dag()
+        if not dag or task_id not in dag.task_ids:
+            return task_id
+        core = re.split(r'__\d+$', task_id)[0]
+        suffixes = sorted(
+            [int(re.split(r'^.+__', task_id)[1])
+             for task_id in dag.task_ids
+             if re.match(rf'^{core}__\d+$', task_id)]
+        )
+        if not suffixes:
+            return f'{core}__1'
+        return f'{core}__{suffixes[-1] + 1}'


We can use f-strings

airflow/operators/python.py

turbaszek · 2020-05-29T11:16:23Z

airflow/operators/python.py

+    """
+    Python operator decorator. Wraps a function into an Airflow operator.
+    Accepts kwargs for operator kwarg. Will try to wrap operator into DAG at declaration or
+    on function invocation. Use alias to reuse function in the DAG.


Is this still valid?

turbaszek · 2020-05-29T11:17:42Z

airflow/operators/python.py

+        kwargs['task_id'] = self._get_unique_task_id(kwargs['task_id'], kwargs.get('dag', None))
+        super().__init__(**kwargs)


Suggested change

kwargs['task_id'] = self._get_unique_task_id(kwargs['task_id'], kwargs.get('dag', None))

super().__init__(**kwargs)

super().__init__(**kwargs)

self.task_id = self._get_unique_task_id(kwargs['task_id'], kwargs.get('dag', None))

Why is this preferred?

No strong opinion here, I just think this is more explicit

Nvm, seems this proposed change fails due to task_id being checked if repeated on super().__init__(**kwargs)

Resolving.

I see another issue here. Currently, if task_id is not provided user will get KeyError: 'task_id' instead of TypeError: __init__() missing 1 required positional argument: 'task_id'

Also, this seems to work as expected:

In [8]: class CustomOp(BaseOperator): ...: def __init__(self, a, b, *args, **kwargs): ...: super().__init__(*args, **kwargs) ...: self.task_id = "other task id" ...: In [9]: op = CustomOp(a=1, b=2, task_id="task_id") In [10]: op.task_id Out[10]: 'other task id'

_PythonFunctionalOperator is a private operator. Aka it should only be used with @task which does always set the task_id.

Will make the field mandatory just in case.

airflow/operators/python.py

turbaszek · 2020-06-02T09:21:27Z

Hi @casassg is there any way we can help you with moving on? 😉

casassg · 2020-06-03T03:27:30Z

Mostly adding more hours to my day. Sorry, been a bit busy this week. Will try to address most comments

turbaszek · 2020-06-10T08:21:05Z

@casassg flake8 is sad, if you wish I can fix it

casassg · 2020-06-10T18:52:21Z

I think I fixed it w latest commit. feel free to push a commit fixing it if it does not. Ran pre-commit locally and it passed

turbaszek · 2020-06-11T09:07:40Z

Two small comments, otherwise it looks really good! I hope we will be able to merge it soon 🚀 It would be good to have another look from someone else. @dimberman @feluelle @kaxil @mik-laj WDYT?

Co-authored-by: Tomek Urbaszek <turbaszek@gmail.com>

Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com>

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com>

Co-authored-by: Felix Uellendall <feluelle@users.noreply.github.com>

casassg · 2020-06-23T18:03:45Z

Rebased from latest master to see if integration tests are fixed.

boring-cyborg · 2020-06-23T21:58:33Z

Awesome work, congrats on your first merged pull request!

kaxil · 2020-06-23T21:58:45Z

Great work @casassg 🎉

dimberman · 2020-06-23T21:59:34Z

This is gonna be awesome! Thank you @casassg !

casassg · 2020-06-23T22:18:38Z

Yay! Thanks everyone for the patience and through review 🎉

potiuk · 2020-06-24T06:20:28Z

Just in time for the Summit !

turbaszek · 2020-06-24T08:45:49Z

@casassg thanks for your work! 🚀

evgenyshulman · 2020-06-24T09:15:26Z

@casassg great initiative and awesome implementation!

Closes apache#8057. Closes apache#8056.

boring-cyborg bot added the area:docs label May 21, 2020

aoen reviewed May 21, 2020

View reviewed changes

casassg mentioned this pull request May 22, 2020

[8057] [AIP-31] Add @task decorator databand-ai/airflow#5

Closed

13 tasks

ashb reviewed May 22, 2020

View reviewed changes

airflow/operators/python.py Show resolved Hide resolved

turbaszek added the AIP-31 Task Flow API for nicer DAG definition label May 22, 2020

turbaszek reviewed May 22, 2020

View reviewed changes

ashb requested changes May 22, 2020

View reviewed changes

evgenyshulman reviewed May 25, 2020

View reviewed changes

airflow/operators/python.py Outdated Show resolved Hide resolved

feluelle reviewed May 27, 2020

View reviewed changes

feluelle reviewed May 28, 2020

View reviewed changes

turbaszek requested changes May 29, 2020

View reviewed changes

casassg force-pushed the feature/task_decorator branch 2 times, most recently from 364243d to ca98528 Compare June 10, 2020 00:38

casassg force-pushed the feature/task_decorator branch from ca98528 to 078c5de Compare June 10, 2020 18:51

turbaszek requested review from feluelle, aoen and ashb June 11, 2020 09:06

turbaszek requested review from mik-laj, dimberman and turbaszek June 11, 2020 09:07

casassg and others added 16 commits June 23, 2020 12:02

Apply suggestions from code review

48d017c

Co-authored-by: Tomek Urbaszek <turbaszek@gmail.com>

added missing tests and updated documentation

5bc9ae4

pylint disables

f43c639

make task_id mandatory

7707b7e

Apply suggestions from code review

e5fe145

Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com>

handle custom task_id

319f1f8

move to TestPythonBase

571ef2a

update documentation

5bb198c

Apply suggestions from code review

9110e6f

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

remove add in 1.10.11 note

adc30d8

fix signature in example

0654425

Apply suggestions from code review

c881f71

Co-authored-by: Ash Berlin-Taylor <ash_github@firemirror.com>

add comment on example dag

c95f416

Apply suggestions from code review

831aca8

Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com>

s/code/code-block

3529296

Update docs/concepts.rst

3dd4113

Co-authored-by: Felix Uellendall <feluelle@users.noreply.github.com>

casassg force-pushed the feature/task_decorator branch from 1c98796 to 3dd4113 Compare June 23, 2020 18:02

ashb approved these changes Jun 23, 2020

View reviewed changes

kaxil merged commit 23faab5 into apache:master Jun 23, 2020

casassg deleted the feature/task_decorator branch June 23, 2020 22:21

kaxil pushed a commit to kaxil/airflow that referenced this pull request Jun 27, 2020

[AIRFLOW-8057] [AIP-31] Add @task decorator (apache#8962)

caec5f8

Closes apache#8057. Closes apache#8056.

casassg mentioned this pull request Aug 18, 2020

infer multiple_output from return type annotation #10349

Merged

		kwargs['task_id'] = self._get_unique_task_id(kwargs['task_id'], kwargs.get('dag', None))
		super().__init__(**kwargs)

[AIRFLOW-8057] [AIP-31] Add @task decorator #8962

[AIRFLOW-8057] [AIP-31] Add @task decorator #8962

Conversation

casassg commented May 21, 2020 • edited Loading

boring-cyborg bot commented May 21, 2020

aoen left a 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

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

casassg May 27, 2020 • edited Loading

Choose a reason for hiding this comment

turbaszek May 28, 2020 • 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

ashb left a 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

ashb commented May 22, 2020

ashb commented May 22, 2020 • edited Loading

casassg commented May 22, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

feluelle May 28, 2020 • 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

casassg commented May 28, 2020

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 Jun 11, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

turbaszek commented Jun 2, 2020

casassg commented Jun 3, 2020

turbaszek commented Jun 10, 2020

casassg commented Jun 10, 2020

turbaszek commented Jun 11, 2020

casassg commented Jun 23, 2020

boring-cyborg bot commented Jun 23, 2020

kaxil commented Jun 23, 2020

dimberman commented Jun 23, 2020

casassg commented Jun 23, 2020

casassg commented May 21, 2020 •

edited

Loading

casassg May 27, 2020 •

edited

Loading

turbaszek May 28, 2020 •

edited

Loading

ashb commented May 22, 2020 •

edited

Loading

feluelle May 28, 2020 •

edited

Loading

turbaszek Jun 11, 2020 •

edited

Loading