This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch airflow-ctl/v0-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/airflow-ctl/v0-1-test by this
push:
new 073142e561e [airflow-ctl/v0-1-test] Keep airflowctl dags
pause/unpause/drain output parseable (#76133) (#76172)
073142e561e is described below
commit 073142e561ee79c120592f67e7ecb1d687e10b05
Author: Shahar Epstein <[email protected]>
AuthorDate: Sat Oct 31 22:57:18 2026 +0200
[airflow-ctl/v0-1-test] Keep airflowctl dags pause/unpause/drain output
parseable (#76133) (#76172)
The "Dag <operation> successful" and "Further Dag details:" status lines,
and the error message on failure, were printed to stdout ahead of the --output
data. That made `airflowctl dags pause <dag_id> -o json` produce output that is
not valid JSON, so piping it to jq or json.loads failed. Status and error text
belong on stderr, matching how the CLI already reports the ambiguous-write
warning, so stdout carries only the requested output format.
Backport note: this branch has no `dags drain`, so the drain cases of the
new tests were dropped; the source change is identical to main.
(cherry picked from commit 956d8cd2d6b19254d86212112e786041439d2a24)
Co-authored-by: Shawn Sun <[email protected]>
---
.../src/airflowctl/ctl/commands/dag_command.py | 7 ++--
.../airflow_ctl/ctl/commands/test_dag_command.py | 47 ++++++++++++++++++++++
2 files changed, 51 insertions(+), 3 deletions(-)
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
index d2acee3d24a..3160277fad3 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
@@ -48,12 +48,13 @@ def update_dag_state(
dag_id=dag_id, dag_body=DAGPatchBody(is_paused=operation ==
"pause")
)
except ServerResponseError as e:
- rich.print(f"[red]Error while trying to {operation} Dag {dag_id}:
{e}[/red]")
+ rich.print(f"[red]Error while trying to {operation} Dag {dag_id}:
{e}[/red]", file=sys.stderr)
sys.exit(1)
response_dict = response.model_dump()
- rich.print(f"[green]Dag {operation} successful {dag_id}[/green]")
- rich.print("[green]Further Dag details:[/green]")
+ # Status lines go to stderr so stdout carries only the --output data (e.g.
JSON for jq).
+ rich.print(f"[green]Dag {operation} successful {dag_id}[/green]",
file=sys.stderr)
+ rich.print("[green]Further Dag details:[/green]", file=sys.stderr)
AirflowConsole().print_as(
data=[response_dict],
output=output,
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
index 62c402afb6c..6153db29677 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
@@ -17,6 +17,7 @@
from __future__ import annotations
import datetime
+import json
from unittest import mock
import httpx
@@ -187,6 +188,52 @@ class TestDagCommands:
api_client=api_client,
)
+ @pytest.mark.parametrize(
+ ("operation", "response_attr"),
+ [
+ ("pause", "dag_response_paused"),
+ ("unpause", "dag_response_unpaused"),
+ ],
+ )
+ def test_update_dag_state_json_output_is_parseable(
+ self, operation, response_attr, api_client_maker, capsys
+ ):
+ dag_response = getattr(self, response_attr)
+ api_client = api_client_maker(
+ path=f"/api/v2/dags/{self.dag_id}",
+ response_json=dag_response.model_dump(mode="json"),
+ expected_http_status_code=200,
+ kind=ClientKind.CLI,
+ )
+ getattr(dag_command, operation)(
+ self.parser.parse_args(["dags", operation, self.dag_id,
"--output", "json"]),
+ api_client=api_client,
+ )
+ captured = capsys.readouterr()
+ assert json.loads(captured.out)[0]["dag_id"] == self.dag_id
+ assert _normalize_rich_output(captured.err) == (
+ f"Dag {operation} successful {self.dag_id} Further Dag details:"
+ )
+
+ @pytest.mark.parametrize("operation", ["pause", "unpause"])
+ def test_update_dag_state_error_goes_to_stderr(self, operation,
api_client_maker, capsys):
+ api_client = api_client_maker(
+ path=f"/api/v2/dags/{self.dag_id}",
+ response_json={"detail": "DAG not found"},
+ expected_http_status_code=404,
+ kind=ClientKind.CLI,
+ )
+ with pytest.raises(SystemExit) as exc_info:
+ getattr(dag_command, operation)(
+ self.parser.parse_args(["dags", operation, self.dag_id,
"--output", "json"]),
+ api_client=api_client,
+ )
+ assert exc_info.value.code == 1
+ captured = capsys.readouterr()
+ error_message = f"Error while trying to {operation} Dag {self.dag_id}"
+ assert error_message in _normalize_rich_output(captured.err)
+ assert error_message not in _normalize_rich_output(captured.out)
+
def test_next_execution(self, api_client_maker):
api_client = api_client_maker(
path=f"/api/v2/dags/{self.dag_id}",
