This is an automated email from the ASF dual-hosted git repository.
potiuk 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 2e4ca56c1e3 fix(airflowctl): handle print for non-output endpoints
(#53785)
2e4ca56c1e3 is described below
commit 2e4ca56c1e3bb947d82ac0e2a7dab5b331b9aacf
Author: Bugra Ozturk <[email protected]>
AuthorDate: Sun Jul 27 10:58:54 2025 +0200
fix(airflowctl): handle print for non-output endpoints (#53785)
---
airflow-ctl/src/airflowctl/ctl/cli_config.py | 33 ++++++++++++++-----
.../tests/airflow_ctl/ctl/test_cli_config.py | 37 +++++++++++++++++++++-
2 files changed, 61 insertions(+), 9 deletions(-)
diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py
b/airflow-ctl/src/airflowctl/ctl/cli_config.py
index 0d22aceacf9..9ef416b0770 100644
--- a/airflow-ctl/src/airflowctl/ctl/cli_config.py
+++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py
@@ -32,6 +32,7 @@ from functools import partial
from pathlib import Path
from typing import Any, NamedTuple
+import httpx
import rich
import airflowctl.api.datamodels.generated as generated_datamodels
@@ -75,6 +76,18 @@ def safe_call_command(function: Callable, args:
Iterable[Arg]) -> None:
except AirflowCtlNotFoundException as e:
rich.print(f"command failed due to {e}")
sys.exit(1)
+ except httpx.RemoteProtocolError as e:
+ if "Server disconnected without sending a response." in str(e):
+ rich.print(
+ f"[red]Server response error: {e}. "
+ "Please check if the server is running and the API URL is
correct.[/red]"
+ )
+ except httpx.ReadTimeout as e:
+ if "timed out" in str(e):
+ rich.print(
+ f"[red]Request timed out: {e}. "
+ "Please check if the server is running and the API ready to
accept calls.[/red]"
+ )
class DefaultHelpParser(argparse.ArgumentParser):
@@ -325,6 +338,7 @@ class CommandFactory:
func_map: dict[tuple, Callable]
commands_map: dict[str, list[ActionCommand]]
group_commands_list: list[CLICommand]
+ ouput_command_list: list[str]
def __init__(self, file_path: str | Path | None = None):
self.datamodels_extended_map = {}
@@ -336,6 +350,8 @@ class CommandFactory:
self.file_path = inspect.getfile(BaseOperations) if file_path is None
else file_path
# Exclude parameters that are not needed for CLI from datamodels
self.excluded_parameters = ["schema_"]
+ # This list is used to determine if the command/operation needs to
output data
+ self.output_command_list = ["list", "get", "create", "delete"]
def _inspect_operations(self) -> None:
"""Parse file and return matching Operation Method with details."""
@@ -524,12 +540,8 @@ class CommandFactory:
parameter_type=parameter_type,
parameter_key=parameter_key
)
)
- # This list is used to determine if the command/operation needs to
output data
- output_command_list = [
- "list",
- "get",
- ]
- if any(operation.get("name").startswith(cmd) for cmd in
output_command_list):
+
+ if any(operation.get("name").startswith(cmd) for cmd in
self.output_command_list):
args.extend([ARG_OUTPUT])
self.args_map[(operation.get("name"),
operation.get("parent").name)] = args
@@ -572,10 +584,13 @@ class CommandFactory:
else:
method_output = operation_method_object(**method_params)
- def convert_to_dict(obj: Any) -> dict | Any:
+ def convert_to_dict(obj: Any, api_operation_name: str) -> dict |
Any:
"""Recursively convert an object to a dictionary or list of
dictionaries."""
if hasattr(obj, "model_dump"):
return obj.model_dump(mode="json")
+ # Handle delete operation which returns a string of the
deleted entity name
+ if isinstance(obj, str):
+ return {"operation": api_operation_name, "entity": obj}
return obj
def check_operation_and_collect_list_of_dict(dict_obj: dict) ->
list:
@@ -598,7 +613,9 @@ class CommandFactory:
return [dict_obj]
AirflowConsole().print_as(
-
data=check_operation_and_collect_list_of_dict(convert_to_dict(method_output)),
+ data=check_operation_and_collect_list_of_dict(
+ convert_to_dict(method_output, api_operation["name"])
+ ),
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 3e887b68431..1e4f5728a25 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
@@ -163,6 +163,30 @@ def test_args_get():
]
[email protected](scope="module")
+def test_args_delete():
+ return [
+ (
+ "--backfill-id",
+ {
+ "help": "backfill_id for delete operation in
BackfillsOperations",
+ "default": None,
+ "type": str,
+ "dest": None,
+ },
+ ),
+ (
+ "--output",
+ {
+ "help": "Output format. Allowed values: json, yaml, plain,
table (default: json)",
+ "default": "json",
+ "type": str,
+ "dest": None,
+ },
+ ),
+ ]
+
+
class TestCommandFactory:
@classmethod
def _save_temp_operations_py(cls, temp_file: str, file_content) -> None:
@@ -183,7 +207,9 @@ class TestCommandFactory:
except FileNotFoundError:
pass
- def test_command_factory(self, no_op_method, test_args_create,
test_args_list, test_args_get):
+ def test_command_factory(
+ self, no_op_method, test_args_create, test_args_list, test_args_get,
test_args_delete
+ ):
"""
Test the command factory.
"""
@@ -213,6 +239,9 @@ class TestCommandFactory:
def get(self, backfill_id: str) -> BackfillResponse |
ServerResponseError:
self.response =
self.client.get(f"backfills/{backfill_id}")
return
BackfillResponse.model_validate_json(self.response.content)
+ def delete(self, backfill_id: str) -> ServerResponseError
| None:
+ self.response =
self.client.delete(f"backfills/{backfill_id}")
+ return None
""",
)
@@ -245,6 +274,12 @@ class TestCommandFactory:
assert arg.kwargs["help"] == test_arg[1]["help"]
assert arg.kwargs["default"] == test_arg[1]["default"]
assert arg.kwargs["type"] == test_arg[1]["type"]
+ elif sub_command.name == "delete":
+ for arg, test_arg in zip(sub_command.args,
test_args_delete):
+ assert arg.flags[0] == test_arg[0]
+ assert arg.kwargs["help"] == test_arg[1]["help"]
+ assert arg.kwargs["default"] == test_arg[1]["default"]
+ assert arg.kwargs["type"] == test_arg[1]["type"]
class TestCliConfigMethods:
