Share airflowctl's Dag run lookup without a new request for run_id

[rjgoyln]

Summary

airflowctl tasks failed-deps and airflowctl tasks states-for-dag-run intentionally do not look up a Dag run when the user supplies a run_id; the ID is passed directly to the task-instance endpoint. That behavior was introduced in #69915 to avoid an unnecessary HTTP request, but the remaining logical-date lookup and selector validation were duplicated across commands.

This PR moves the shared logic into airflowctl/ctl/utils/dag_run.py, removing the duplication while preserving the existing request pattern and user-visible behavior.

Why two resolvers?

The split is intentional.

• resolve_dag_run_id returns a supplied run_id directly, avoiding the extra Dag run lookup.
• resolve_dag_run fetches the Dag run because commands such as dags state need the full response to display fields such as state and conf.

Both helpers share the selector validation and logical-date lookup, but keep the different run_id behavior intact.

A single parameterized resolver was considered, but since the two helpers naturally return different types (str vs DAGRunResponse), combining them would require either a union return type or @overload definitions. That adds more complexity than the small amount of dispatch logic it would eliminate.

Tests

The "don't look up a supplied run_id" behavior was previously not enforced by the test suite.

As a demonstration, adding a verification-only dag_runs.get call (for example, to validate a supplied run_id) produces the following result:

Branch	Result
main	30/30 tests pass
This branch	Exactly 2 tests fail (the assertions added here)

The new assertions make this behavior an explicit contract and prevent future refactors from unintentionally adding the extra request.

Notes

• No behavior change is intended. Request counts, error messages, and exit paths remain unchanged.
• No newsfragment: airflow-ctl release managers regenerate the changelog from git log.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

[SameerMesiah97]

Looks good. This is a relatively low-risk refactor,

[potiuk]

Heads-up on an incoming conflict — not a review.

@haseebmalik18's #71206 adds an airflowctl tasks state command, and it brings a third copy of the block this PR consolidates:

if (args.run_id is None) == (args.logical_date is None):
    rich.print("[red]Provide either run_id or --logical-date, but not both[/red]")
    sys.exit(1)
run_id = args.run_id or _find_run_id_by_logical_date(api_client, args.dag_id, args.logical_date)

I've approved that PR, so it will most likely land first. When it does, this one will conflict in task_command.py, and the resolution is small: the new state command's block collapses into another resolve_dag_run_id(api_client, args) call, next to the two in failed_deps and states_for_dag_run.

Nothing to do right now. Flagging it so the conflict isn't a surprise when it turns up, and so the extra call site reads as expected rather than as something that crept in while this was waiting.

This note is only about the overlap — I haven't left a review on this PR itself.

---

This message was drafted by an AI-assisted tool and confirmed by an Airflow maintainer. If anything here looks wrong, please reply on the PR and a maintainer will follow up.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[potiuk]

Clean refactor — the two lookup helpers move verbatim, the duplicated logical-date lookup in task_command.py collapses into the shared one, and keeping resolve_dag_run / resolve_dag_run_id separate (with the docstring saying why a supplied run_id isn't fetched) is the right call. The new assert_not_called() checks turn that "no extra request" behaviour into an enforced contract. Approving.

---

This review was drafted by an AI-assisted tool and
confirmed by an Apache Airflow maintainer. The maintainer
approving this PR has read the findings and signed off. If
something feels off, please reply on the PR and a maintainer
will follow up.

More on how Apache Airflow handles maintainer review:
contributing-docs/05_pull_requests.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[github-actions]

Backport failed to create: airflow-ctl/v0-1-test. View the failure log Run details

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
❌	airflow-ctl/v0-1-test

You can attempt to backport this manually by running:

cherry_picker 6e48e0f airflow-ctl/v0-1-test

This should apply the commit to the airflow-ctl/v0-1-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.