This is an automated email from the ASF dual-hosted git repository.
henry3260 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 ddb905f0ee0 Fix airflowctl dags get-tags crashing whenever a Dag has
tags (#73743)
ddb905f0ee0 is described below
commit ddb905f0ee09b423d0b09ef79e62c282de9eb14c
Author: Stefan Wang <[email protected]>
AuthorDate: Wed Oct 7 21:43:01 2026 -0700
Fix airflowctl dags get-tags crashing whenever a Dag has tags (#73743)
* Fix airflowctl dags get-tags crashing whenever a Dag has tags
The dagTags endpoint returns tag names as plain strings, but the printer
behind every generated command only renders records. The command raised
"To tabulate non-dictionary data you need to provide `mapper` function"
for every output format. Print one row per tag instead.
Signed-off-by: 1fanwang <[email protected]>
* Pin airflowctl outputs that keep their single-key shape
Cover pools list printing each pool as a record and dagrun get keeping a
single-key list conf on one row, so dropping either guard fails a test.
Signed-off-by: 1fanwang <[email protected]>
---------
Signed-off-by: 1fanwang <[email protected]>
---
.../airflowctl_tests/test_airflowctl_commands.py | 1 +
airflow-ctl/src/airflowctl/ctl/cli_config.py | 12 ++-
.../tests/airflow_ctl/ctl/test_cli_config.py | 96 ++++++++++++++++++++++
3 files changed, 105 insertions(+), 4 deletions(-)
diff --git
a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
index 377a5d8a87d..0e7ce34ed8f 100644
--- a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
+++ b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
@@ -84,6 +84,7 @@ TEST_COMMANDS = [
"dags get example_bash_operator",
"dags get-details example_bash_operator",
"dags get-stats example_bash_operator",
+ "dags get-tags",
"dags get-version example_bash_operator 1",
"dags list-import-errors",
"dags list-version example_bash_operator",
diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py
b/airflow-ctl/src/airflowctl/ctl/cli_config.py
index 725e75be8d6..d160772eef3 100755
--- a/airflow-ctl/src/airflowctl/ctl/cli_config.py
+++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py
@@ -40,7 +40,7 @@ import rich
import airflowctl.api.datamodels.generated as generated_datamodels
from airflowctl.api.client import NEW_API_CLIENT, Client, ClientKind,
provide_api_client
from airflowctl.api.operations import BaseOperations, ServerResponseError
-from airflowctl.ctl.console_formatting import AirflowConsole
+from airflowctl.ctl.console_formatting import AirflowConsole, is_data_sequence
from airflowctl.ctl.utils.yaml import safe_load
from airflowctl.exceptions import (
AirflowCtlConnectionException,
@@ -905,7 +905,7 @@ class CommandFactory:
return {"operation": api_operation_name, "entity": obj}
return obj
- def check_operation_and_collect_list_of_dict(dict_obj: dict) ->
list:
+ def check_operation_and_collect_list_of_dict(dict_obj: dict,
top_level: bool = False) -> list:
"""Check if the object is a nested dictionary and collect list
of dictionaries."""
def is_dict_nested(obj: dict) -> bool:
@@ -928,13 +928,17 @@ class CommandFactory:
# If dict_obj only have single key return value instead of list
# This can happen since we are excluding some keys from user
such as total_entries from list operations
if len(dict_obj) == 1:
- return dict_obj[next(iter(dict_obj.keys()))]
+ key, value = next(iter(dict_obj.items()))
+ # Printed rows must be records, so plain values such as
Dag tags get one row each.
+ if top_level and isinstance(value, list) and not
is_data_sequence(value):
+ return [{key: item} for item in value]
+ return value
# If not nested, return the object as a list which the result
should be already a dict
return [dict_obj]
AirflowConsole().print_as(
data=check_operation_and_collect_list_of_dict(
- convert_to_dict(method_output, api_operation["name"])
+ dict_obj=convert_to_dict(method_output,
api_operation["name"]), top_level=True
),
output=args.output,
)
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
index f839e4e0970..d092b579570 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
@@ -18,10 +18,12 @@
from __future__ import annotations
import argparse
+import json
from argparse import BooleanOptionalAction
from datetime import UTC
from pathlib import Path
from textwrap import dedent
+from typing import Any
from unittest import mock
import httpx
@@ -44,6 +46,7 @@ from airflowctl.ctl.cli_config import (
safe_call_command,
)
from airflowctl.ctl.console_formatting import AirflowConsole
+from airflowctl.ctl.utils.yaml import safe_load
from airflowctl.exceptions import (
AirflowCtlConnectionException,
AirflowCtlCredentialNotFoundException,
@@ -988,3 +991,96 @@ class TestCliConfigMethods:
args.func(args, api_client=mock.MagicMock(spec=Client))
assert mocked_print_as.call_args.kwargs["output"] == "json"
+
+ @pytest.mark.parametrize(
+ ("output", "parse", "expected"),
+ [
+ pytest.param("json", json.loads, [{"tags": "etl"}, {"tags":
"nightly"}], id="json"),
+ pytest.param("yaml", safe_load, [{"tags": "etl"}, {"tags":
"nightly"}], id="yaml"),
+ pytest.param("table", str.split, ["tags", "=======", "etl",
"nightly"], id="table"),
+ pytest.param("plain", str.split, ["tags", "etl", "nightly"],
id="plain"),
+ ],
+ )
+ def test_collection_of_plain_values_prints_one_row_per_value(
+ self, api_client_maker, capsys, output, parse, expected
+ ):
+ """``dags get-tags`` receives plain strings, and the printer only
renders records."""
+ api_client = api_client_maker(
+ path="/api/v2/dagTags",
+ response_json={"tags": ["etl", "nightly"], "total_entries": 2},
+ expected_http_status_code=200,
+ )
+ args = cli_parser.get_parser().parse_args(["dags", "get-tags",
"--output", output])
+
+ args.func(args, api_client=api_client)
+
+ assert parse(capsys.readouterr().out) == expected
+
+ def test_list_of_records_prints_records_unwrapped(self, api_client_maker,
capsys):
+ pools: list[dict[str, Any]] = [
+ {
+ "name": name,
+ "slots": 128,
+ "description": None,
+ "include_deferred": False,
+ "occupied_slots": 0,
+ "running_slots": 0,
+ "queued_slots": 0,
+ "scheduled_slots": 0,
+ "open_slots": 128,
+ "deferred_slots": 0,
+ "team_name": None,
+ }
+ for name in ("default_pool", "etl")
+ ]
+ api_client = api_client_maker(
+ path="/api/v2/pools",
+ response_json={"pools": pools, "total_entries": 2},
+ expected_http_status_code=200,
+ )
+ args = cli_parser.get_parser().parse_args(["pools", "list",
"--output", "json"])
+
+ args.func(args, api_client=api_client)
+
+ rows = json.loads(capsys.readouterr().out)
+ assert [row["name"] for row in rows] == ["default_pool", "etl"]
+ assert all(row.keys() == pools[0].keys() for row in rows)
+
+ def test_nested_single_key_list_is_not_split_into_rows(self,
api_client_maker, capsys):
+ api_client = api_client_maker(
+ path="/api/v2/dags/example/dagRuns/manual_run",
+ response_json={
+ "dag_run_id": "manual_run",
+ "dag_id": "example",
+ "logical_date": None,
+ "queued_at": None,
+ "start_date": None,
+ "end_date": None,
+ "duration": None,
+ "data_interval_start": None,
+ "data_interval_end": None,
+ "run_after": "2025-01-01T00:00:00Z",
+ "last_scheduling_decision": None,
+ "run_type": "manual",
+ "state": "success",
+ "triggered_by": None,
+ "triggering_user_name": None,
+ "conf": {"ids": ["a", "b"]},
+ "note": None,
+ "dag_versions": [],
+ "bundle_version": None,
+ "dag_display_name": "example",
+ "partition_key": None,
+ "partition_date": None,
+ },
+ expected_http_status_code=200,
+ )
+ args = cli_parser.get_parser().parse_args(
+ ["dagrun", "get", "example", "manual_run", "--output", "json"]
+ )
+
+ args.func(args, api_client=api_client)
+
+ rows = json.loads(capsys.readouterr().out)
+ assert [row["dag_run_id"] for row in rows] == ["manual_run"]
+ assert rows[0]["conf"] == ["a", "b"]
