Skip to content

Commit

Permalink
refactor(models/taskinstance): rename _defer_task as defer_task (stil…
Browse files Browse the repository at this point in the history
…l mark it as private)
  • Loading branch information
Lee-W committed Feb 21, 2024
1 parent 1d0020c commit 7b55586
Show file tree
Hide file tree
Showing 2 changed files with 7 additions and 4 deletions.
2 changes: 1 addition & 1 deletion airflow/cli/commands/task_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -636,7 +636,7 @@ def task_test(args, dag: DAG | None = None, session: Session = NEW_SESSION) -> N
else:
ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True, raise_on_defer=True)
except TaskDeferred as defer:
ti._defer_task(defer=defer, session=session)
ti.defer_task(defer=defer, session=session)
log.info("[TASK TEST] running trigger in line")

event = _run_inline_trigger(defer.trigger)
Expand Down
9 changes: 6 additions & 3 deletions airflow/models/taskinstance.py
Original file line number Diff line number Diff line change
Expand Up @@ -2378,7 +2378,7 @@ def _run_raw_task(
# a trigger.
if raise_on_defer:
raise
self._defer_task(defer=defer, session=session)
self.defer_task(defer=defer, session=session)
self.log.info(
"Pausing task as DEFERRED. dag_id=%s, task_id=%s, execution_date=%s, start_date=%s",
self.dag_id,
Expand Down Expand Up @@ -2565,8 +2565,11 @@ def _execute_task(self, context, task_orig):
return _execute_task(self, context, task_orig)

@provide_session
def _defer_task(self, session: Session, defer: TaskDeferred) -> None:
"""Mark the task as deferred and sets up the trigger that is needed to resume it."""
def defer_task(self, session: Session, defer: TaskDeferred) -> None:
"""Mark the task as deferred and sets up the trigger that is needed to resume it.
:meta: private
"""
from airflow.models.trigger import Trigger

# First, make the trigger entry
Expand Down

0 comments on commit 7b55586

Please sign in to comment.