Skip to content

Commit 16ad479

Browse files
Add dags next execution command #66172 (#66188)
* Add airflowctl dags next-execution command #66172 * Add generated OpenAPI spec and UI types * Revert "Add generated OpenAPI spec and UI types" This reverts commit 6748ed8. * Update help text Dag definition --------- Co-authored-by: bugraoz93 <ozturkbugra93@gmail.com>
1 parent 336a119 commit 16ad479

6 files changed

Lines changed: 179 additions & 62 deletions

File tree

airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@ def date_param():
8686
"dags trigger example_bash_operator --logical-date={date_param} --run-after={date_param}",
8787
# Test trigger without logical-date (should default to now)
8888
"dags trigger example_bash_operator",
89+
"dags next-execution example_bash_operator",
8990
"dags pause example_bash_operator",
9091
"dags unpause example_bash_operator",
9192
# Dag Run commands

airflow-ctl/docs/images/command_hashes.txt

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ auth:d79e9c7d00c432bdbcbc2a86e2e32053
44
backfill:74c8737b0a62a86ed3605fa9e6165874
55
config:a3d936cb15fe3b547bf6c82cf93d923f
66
connections:942f9f88cb908c28bf5c19159fc5065b
7-
dags:e2a18f90b1bd150be981cef6fef91858
7+
dags:6b38e6bcd491bc1941e7814b77e63bde
88
dagrun:c32e0011aa9a845456c778786717208e
99
jobs:a5b644c5da8889443bb40ee10b599270
1010
pools:19efe105b9515ab1926ebcaf0e028d71

airflow-ctl/docs/images/output_dags.svg

Lines changed: 65 additions & 61 deletions
Loading

airflow-ctl/src/airflowctl/ctl/cli_config.py

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -959,6 +959,15 @@ def merge_commands(
959959
)
960960

961961
DAG_COMMANDS = (
962+
ActionCommand(
963+
name="next-execution",
964+
help="Show the next scheduled execution time for a Dag",
965+
func=lazy_load_command("airflowctl.ctl.commands.dag_command.next_execution"),
966+
args=(
967+
ARG_DAG_ID,
968+
ARG_OUTPUT,
969+
),
970+
),
962971
ActionCommand(
963972
name="pause",
964973
help="Pause a Dag",

airflow-ctl/src/airflowctl/ctl/commands/dag_command.py

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,3 +72,34 @@ def unpause(args, api_client=NEW_API_CLIENT) -> None:
7272
api_client=api_client,
7373
output=args.output,
7474
)
75+
76+
77+
_NEXT_EXECUTION_FIELDS = (
78+
"next_dagrun_logical_date",
79+
"next_dagrun_data_interval_start",
80+
"next_dagrun_data_interval_end",
81+
"next_dagrun_run_after",
82+
)
83+
84+
85+
@provide_api_client(kind=ClientKind.CLI)
86+
def next_execution(args, api_client=NEW_API_CLIENT) -> dict | None:
87+
"""Show next scheduled execution time for a DAG."""
88+
try:
89+
response = api_client.dags.get(dag_id=args.dag_id)
90+
except ServerResponseError as e:
91+
rich.print(f"[red]Error retrieving DAG {args.dag_id}: {e}[/red]")
92+
sys.exit(1)
93+
94+
next_exec_data = {field: getattr(response, field) for field in _NEXT_EXECUTION_FIELDS}
95+
96+
if all(value is None for value in next_exec_data.values()):
97+
rich.print(f"[yellow]No upcoming run scheduled for DAG {args.dag_id}.[/yellow]")
98+
return None
99+
100+
result = next_exec_data
101+
AirflowConsole().print_as(
102+
data=[result],
103+
output=args.output,
104+
)
105+
return result

airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,36 @@ class TestDagCommands:
9090
is_stale=False,
9191
)
9292

93+
dag_response_no_schedule = DAGResponse(
94+
dag_id=dag_id,
95+
dag_display_name=dag_display_name,
96+
is_paused=True,
97+
last_parsed_time=datetime.datetime(2024, 12, 31, 23, 59, 59),
98+
last_expired=datetime.datetime(2025, 1, 1, 0, 0, 0),
99+
fileloc="fileloc",
100+
relative_fileloc="relative_fileloc",
101+
description="description",
102+
timetable_summary=None,
103+
timetable_description=None,
104+
timetable_partitioned=False,
105+
timetable_periodic=False,
106+
tags=[],
107+
max_active_tasks=1,
108+
max_active_runs=1,
109+
max_consecutive_failed_dag_runs=1,
110+
has_task_concurrency_limits=False,
111+
has_import_errors=False,
112+
next_dagrun_logical_date=None,
113+
next_dagrun_data_interval_start=None,
114+
next_dagrun_data_interval_end=None,
115+
next_dagrun_run_after=None,
116+
owners=["apache-airflow"],
117+
is_backfillable=False,
118+
file_token="file_token",
119+
bundle_name="bundle_name",
120+
is_stale=False,
121+
)
122+
93123
def test_pause_dag(self, api_client_maker, monkeypatch):
94124
api_client = api_client_maker(
95125
path=f"/api/v2/dags/{self.dag_id}",
@@ -143,3 +173,45 @@ def test_unpause_fail(self, api_client_maker, monkeypatch):
143173
self.parser.parse_args(["dags", "unpause", self.dag_id]),
144174
api_client=api_client,
145175
)
176+
177+
def test_next_execution(self, api_client_maker):
178+
api_client = api_client_maker(
179+
path=f"/api/v2/dags/{self.dag_id}",
180+
response_json=self.dag_response_paused.model_dump(mode="json"),
181+
expected_http_status_code=200,
182+
kind=ClientKind.CLI,
183+
)
184+
result = dag_command.next_execution(
185+
self.parser.parse_args(["dags", "next-execution", self.dag_id]),
186+
api_client=api_client,
187+
)
188+
assert result["next_dagrun_logical_date"] == datetime.datetime(2025, 1, 1, 0, 0, 0)
189+
assert result["next_dagrun_data_interval_start"] == datetime.datetime(2025, 1, 1, 0, 0, 0)
190+
assert result["next_dagrun_data_interval_end"] == datetime.datetime(2025, 1, 1, 0, 0, 0)
191+
assert result["next_dagrun_run_after"] == datetime.datetime(2025, 1, 1, 0, 0, 0)
192+
193+
def test_next_execution_no_schedule(self, api_client_maker):
194+
api_client = api_client_maker(
195+
path=f"/api/v2/dags/{self.dag_id}",
196+
response_json=self.dag_response_no_schedule.model_dump(mode="json"),
197+
expected_http_status_code=200,
198+
kind=ClientKind.CLI,
199+
)
200+
result = dag_command.next_execution(
201+
self.parser.parse_args(["dags", "next-execution", self.dag_id]),
202+
api_client=api_client,
203+
)
204+
assert result is None
205+
206+
def test_next_execution_fail(self, api_client_maker):
207+
api_client = api_client_maker(
208+
path=f"/api/v2/dags/{self.dag_id}",
209+
response_json={"detail": "DAG not found"},
210+
expected_http_status_code=404,
211+
kind=ClientKind.CLI,
212+
)
213+
with pytest.raises(SystemExit):
214+
dag_command.next_execution(
215+
self.parser.parse_args(["dags", "next-execution", self.dag_id]),
216+
api_client=api_client,
217+
)

0 commit comments

Comments
 (0)