This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 956d8cd2d6b Keep airflowctl dags pause/unpause/drain output parseable
(#76133)
956d8cd2d6b is described below
commit 956d8cd2d6b19254d86212112e786041439d2a24
Author: Shawn Sun <[email protected]>
AuthorDate: Sat Oct 31 05:59:22 2026 -0700
Keep airflowctl dags pause/unpause/drain output parseable (#76133)
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.
---
.../src/airflowctl/ctl/commands/dag_command.py | 7 ++--
.../airflow_ctl/ctl/commands/test_dag_command.py | 48 ++++++++++++++++++++++
2 files changed, 52 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 95770afa81d..2dc051769bd 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
@@ -55,12 +55,13 @@ def update_dag_state(
try:
response = api_client.dags.update(dag_id=dag_id, dag_body=dag_body)
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 a11b83adee2..09f6b3b43d4 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 types import SimpleNamespace
from unittest import mock
from unittest.mock import Mock, call, create_autospec
@@ -280,6 +281,53 @@ class TestDagCommands:
api_client=api_client,
)
+ @pytest.mark.parametrize(
+ ("operation", "response_attr"),
+ [
+ ("pause", "dag_response_paused"),
+ ("unpause", "dag_response_unpaused"),
+ ("drain", "dag_response_draining"),
+ ],
+ )
+ 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", "drain"])
+ 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}",
