Skip to content

Commit

Permalink
Determine needs_expansion on client side when DB isolated
Browse files Browse the repository at this point in the history
This is necessary because Operator does not serialize quite right.  It doesn't (either completely or at all) include the dag object or its groups.  So we can't properly determine 'needs expansion' on server side when it's in a mapped task group.
  • Loading branch information
dstandish committed Apr 27, 2024
1 parent b977335 commit baf7079
Showing 1 changed file with 6 additions and 2 deletions.
8 changes: 6 additions & 2 deletions airflow/cli/commands/task_command.py
Expand Up @@ -166,6 +166,7 @@ def _get_ti_db_access(
exec_date_or_run_id: str | None = None,
pool: str | None = None,
create_if_necessary: CreateIfNecessary = False,
should_expand=False,
session: Session = NEW_SESSION,
) -> tuple[TaskInstance | TaskInstancePydantic, bool]:
"""Get the task instance through DagRun.run_id, if that fails, get the TI the old way."""
Expand All @@ -177,7 +178,7 @@ def _get_ti_db_access(

if not exec_date_or_run_id and not create_if_necessary:
raise ValueError("Must provide `exec_date_or_run_id` if not `create_if_necessary`.")
if needs_expansion(task):
if should_expand:
if map_index < 0:
raise RuntimeError("No map_index passed to mapped task")
elif map_index >= 0:
Expand Down Expand Up @@ -218,18 +219,21 @@ def _get_ti(
if dag is None:
raise ValueError("Cannot get task instance for a task not assigned to a DAG")

should_expand = needs_expansion(task)
ti, dr_created = _get_ti_db_access(
dag=dag,
task=task,
map_index=map_index,
exec_date_or_run_id=exec_date_or_run_id,
pool=pool,
create_if_necessary=create_if_necessary,
should_expand=should_expand,
)

# setting ti.task is necessary for AIP-44 since the task object does not serialize perfectly
# if we update the serialization logic for Operator to also serialize the dag object on it,
# then this would not be necessary;
ti.task = task
ti.refresh_from_task(task, pool_override=pool)
return ti, dr_created


Expand Down

0 comments on commit baf7079

Please sign in to comment.