Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Classic TaskGroup setup/teardown #29891

Merged
merged 2 commits into from
Mar 14, 2023
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
2 changes: 2 additions & 0 deletions airflow/decorators/task_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,8 @@ def task_group(
ui_color: str = "CornflowerBlue",
ui_fgcolor: str = "#000",
add_suffix_on_collision: bool = False,
setup: bool = False,
teardown: bool = False,
) -> Callable[[Callable[FParams, FReturn]], _TaskGroupFactory[FParams, FReturn]]:
...

Expand Down
33 changes: 31 additions & 2 deletions airflow/utils/task_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,18 @@ def __init__(
ui_color: str = "CornflowerBlue",
ui_fgcolor: str = "#000",
add_suffix_on_collision: bool = False,
setup: bool = False,
teardown: bool = False,
):
from airflow.models.dag import DagContext

if setup and teardown:
raise AirflowException("Cannot set both setup and teardown to True")

self.prefix_group_id = prefix_group_id
self.default_args = copy.deepcopy(default_args or {})
self.setup = setup
self.teardown = teardown
ephraimbuddy marked this conversation as resolved.
Show resolved Hide resolved

dag = dag or DagContext.get_current_dag()

Expand Down Expand Up @@ -231,15 +238,37 @@ def add(self, task: DAGNode) -> None:
if task.children:
raise AirflowException("Cannot add a non-empty TaskGroup")

if SetupTeardownContext.is_setup:
is_setup, is_teardown = self._check_is_setup_teardown(task)

if SetupTeardownContext.is_setup or is_setup:
if isinstance(task, AbstractOperator):
setattr(task, "_is_setup", True)
elif SetupTeardownContext.is_teardown:
elif SetupTeardownContext.is_teardown or is_teardown:
if isinstance(task, AbstractOperator):
setattr(task, "_is_teardown", True)

self.children[key] = task

def _check_is_setup_teardown(self, task_):
"""Check if setup or teardown is set for the task"""
from airflow.models.abstractoperator import AbstractOperator

def _find_setup_teardown(tg):
setup, teardown = tg.setup, tg.teardown
while tg and tg.parent_group:
if setup or teardown:
break
tg = tg.parent_group
setup, teardown = tg.setup, tg.teardown
return setup, teardown

if isinstance(task_, TaskGroup):
return _find_setup_teardown(task_)
if isinstance(task_, AbstractOperator):
tg = task_.task_group
return _find_setup_teardown(tg)
return False, False

def _remove(self, task: DAGNode) -> None:
key = task.node_id

Expand Down
Loading