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 9cde6aa11db Stop airflowctl updates from resetting omitted bool fields
(#74383)
9cde6aa11db is described below
commit 9cde6aa11dbdd8dfaba8119a65fe768215153141
Author: Shawn Sun <[email protected]>
AuthorDate: Sun Nov 1 01:47:45 2026 -0800
Stop airflowctl updates from resetting omitted bool fields (#74383)
* Stop airflowctl updates from resetting omitted bool fields
Squashed onto main for a clean rebase. Includes the integration-test and
backfill test changes from Shahar Epstein's commit 82a7ed8e0c.
Generated-by: GitHub Copilot CLI (Claude Opus 5.5)
* Fail airflowctl pools update early when --include-deferred is omitted
The server requires include_deferred on a pool PATCH without an update
mask, so an omitted flag now gets the same client-side message as dags update
instead of a raw 422.
Generated-by: Claude Opus 5
---------
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../airflowctl_tests/test_airflowctl_commands.py | 2 +-
airflow-ctl/src/airflowctl/api/operations.py | 2 +-
airflow-ctl/src/airflowctl/ctl/cli_config.py | 35 +++---
.../tests/airflow_ctl/api/test_operations.py | 9 ++
.../tests/airflow_ctl/ctl/test_cli_config.py | 122 ++++++++++++++++++++-
5 files changed, 154 insertions(+), 16 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 79ed6d72229..48a213902d4 100644
--- a/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
+++ b/airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py
@@ -138,7 +138,7 @@ TEST_COMMANDS = [
"pools list",
"pools get test_pool",
"pools get test_pool -o yaml",
- "pools update --pool=test_pool --slots=10",
+ "pools update --pool=test_pool --slots=10 --no-include-deferred",
"pools import tests/airflowctl_tests/fixtures/test_pools.json",
"pools export tests/airflowctl_tests/fixtures/pools_export.json
--output=json",
"pools delete test_pool",
diff --git a/airflow-ctl/src/airflowctl/api/operations.py
b/airflow-ctl/src/airflowctl/api/operations.py
index 497bec2f11a..556d857f371 100644
--- a/airflow-ctl/src/airflowctl/api/operations.py
+++ b/airflow-ctl/src/airflowctl/api/operations.py
@@ -616,7 +616,7 @@ class DagsOperations(BaseOperations):
def update(self, dag_id: str, dag_body: DAGPatchBody) -> DAGResponse |
ServerResponseError:
self.response = self.client.patch(
- f"dags/{_quote_path_param(dag_id)}",
json=dag_body.model_dump(mode="json")
+ f"dags/{_quote_path_param(dag_id)}",
json=dag_body.model_dump(mode="json", exclude_none=True)
)
return validate_response(self.response.content, DAGResponse)
diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py
b/airflow-ctl/src/airflowctl/ctl/cli_config.py
index 9ebabc729e0..455454fc271 100755
--- a/airflow-ctl/src/airflowctl/ctl/cli_config.py
+++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py
@@ -603,10 +603,6 @@ class CommandFactory:
"set",
"clear",
]
- # Datamodels whose generated bool flags follow the datamodel field
defaults instead of
- # defaulting to False, so the CLI keeps the API semantics (e.g. a bare
``tasks clear``
- # must keep ``dry_run=True`` and preview instead of clearing).
- self.field_bool_default_datamodels = ["ClearTaskInstancesBody"]
self.exclude_operation_names = ["LoginOperations",
"VersionOperations", "BaseOperations"]
self.exclude_method_names = [
"error",
@@ -804,11 +800,16 @@ class CommandFactory:
return suffix
return None
- def _get_bool_arg_default(self, parameter_type: str, field_default: Any)
-> bool | None:
- """Get default for a generated bool flag: the datamodel field default
for datamodels in ``field_bool_default_datamodels``, otherwise False."""
- if parameter_type in self.field_bool_default_datamodels and (
- field_default is None or isinstance(field_default, bool)
- ):
+ @staticmethod
+ def _get_bool_arg_default(field_default: Any) -> bool | None:
+ """
+ Get the default for a generated bool flag from its datamodel field
default.
+
+ An omitted flag must keep the API semantics: a bare ``tasks clear``
keeps ``dry_run=True``,
+ and an optional field such as ``is_paused`` or ``include_deferred``
stays ``None`` so it is
+ not sent at all instead of overwriting the stored value with ``False``.
+ """
+ if field_default is None or isinstance(field_default, bool):
return field_default
return False
@@ -841,9 +842,7 @@ class CommandFactory:
arg_action=argparse.BooleanOptionalAction if annotation is
bool else None, # type: ignore
arg_help=f"{field} for {parameter_key} operation",
arg_default=(
- self._get_bool_arg_default(parameter_type,
field_type.default)
- if annotation is bool
- else None
+ self._get_bool_arg_default(field_type.default) if
annotation is bool else None
),
arg_required=(
field_type.is_required() # no default?
@@ -911,7 +910,7 @@ class CommandFactory:
def _apply_datamodel_defaults(self, datamodel: type, params: dict) -> dict:
"""
- Apply datamodel-specific default values.
+ Apply datamodel-specific default values and checks.
Centralizes special handling for different datamodels to keep
CLI parameter processing logic clean and maintainable.
@@ -950,6 +949,16 @@ class CommandFactory:
else:
params[field] = [item.strip() for item in
raw_value.split(",") if item.strip()]
+ # The server requires exactly one of these, so fail early instead of
showing its raw validation error
+ if datamodel.__name__ == "DAGPatchBody":
+ provided = [field for field in ("is_paused", "scheduling_state")
if params.get(field) is not None]
+ if len(provided) != 1:
+ raise SystemExit("Provide exactly one of --is-paused /
--no-is-paused or --scheduling-state.")
+
+ # The server requires include_deferred on a pool PATCH without an
update mask
+ if datamodel.__name__ == "PoolPatchBody" and
params.get("include_deferred") is None:
+ raise SystemExit("Provide --include-deferred or
--no-include-deferred.")
+
return params
def _validate_datamodel(self, datamodel: type[BaseModel], params: dict) ->
Any:
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index debac6ccacb..b7efd5b02a6 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -1496,6 +1496,15 @@ class TestDagOperations:
response = client.dags.update(dag_id="dag_id",
dag_body=self.dag_patch_body)
assert response == self.dag_response
+ def test_update_omits_none_fields_from_request_body(self):
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert json.loads(request.content.decode()) == {"is_paused": True}
+ return httpx.Response(200,
json=json.loads(self.dag_response.model_dump_json()))
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.dags.update(dag_id="dag_id",
dag_body=self.dag_patch_body)
+ assert response == self.dag_response
+
def test_delete(self):
def handle_request(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/api/v2/dags/dag_id"
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 93ead45c2f7..3d09bef569e 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py
@@ -32,14 +32,25 @@ import pytest
from airflowctl.api.client import Client
from airflowctl.api.datamodels.generated import (
+ BackfillPostBody,
ClearTaskInstancesBody,
ConnectionBody,
ConnectionTestResponse,
+ DAGPatchBody,
+ DagSchedulingState,
PoolBody,
PoolCollectionResponse,
+ PoolPatchBody,
PoolResponse,
)
-from airflowctl.api.operations import ConnectionsOperations, DagRunOperations,
ServerResponseError
+from airflowctl.api.operations import (
+ BackfillOperations,
+ ConnectionsOperations,
+ DagRunOperations,
+ DagsOperations,
+ PoolsOperations,
+ ServerResponseError,
+)
from airflowctl.ctl import cli_parser
from airflowctl.ctl.cli_config import (
ARG_AUTH_TOKEN,
@@ -1494,6 +1505,115 @@ class TestCliConfigMethods:
assert [row["dag_run_id"] for row in rows] == ["manual_run"]
assert rows[0]["conf"] == ["a", "b"]
+ @pytest.mark.parametrize(
+ ("extra_argv", "expected_body"),
+ [
+ pytest.param(
+ ["--scheduling-state", "paused"],
+ DAGPatchBody(scheduling_state=DagSchedulingState.PAUSED),
+ id="scheduling-state-alone",
+ ),
+ pytest.param(["--is-paused"], DAGPatchBody(is_paused=True),
id="explicit-pause"),
+ pytest.param(["--no-is-paused"], DAGPatchBody(is_paused=False),
id="explicit-unpause"),
+ ],
+ )
+ @mock.patch.object(AirflowConsole, "print_as", autospec=True)
+ @mock.patch.object(DagsOperations, "update", autospec=True)
+ def test_dags_update_sends_only_given_state(
+ self, mocked_update, mocked_print_as, extra_argv, expected_body
+ ):
+ args = cli_parser.get_parser().parse_args(["dags", "update", "my_dag",
*extra_argv])
+
+ args.func(args, api_client=mock.MagicMock(spec=Client))
+
+ assert mocked_update.call_args.kwargs["dag_body"] == expected_body
+
+ @pytest.mark.parametrize(
+ "extra_argv",
+ [
+ pytest.param([], id="no-state-does-not-unpause"),
+ pytest.param(["--is-paused", "--scheduling-state", "paused"],
id="both-states"),
+ ],
+ )
+ @mock.patch.object(AirflowConsole, "print_as", autospec=True)
+ @mock.patch.object(DagsOperations, "update", autospec=True)
+ def test_dags_update_requires_exactly_one_state(self, mocked_update,
mocked_print_as, extra_argv):
+ args = cli_parser.get_parser().parse_args(["dags", "update", "my_dag",
*extra_argv])
+
+ with pytest.raises(
+ SystemExit, match="exactly one of --is-paused / --no-is-paused or
--scheduling-state"
+ ):
+ args.func(args, api_client=mock.MagicMock(spec=Client))
+
+ mocked_update.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("extra_argv", "expected_include_deferred"),
+ [
+ pytest.param(["--include-deferred"], True, id="explicit-include"),
+ pytest.param(["--no-include-deferred"], False,
id="explicit-exclude"),
+ ],
+ )
+ @mock.patch.object(AirflowConsole, "print_as", autospec=True)
+ @mock.patch.object(PoolsOperations, "update", autospec=True)
+ def test_pools_update_sends_given_include_deferred(
+ self, mocked_update, mocked_print_as, extra_argv,
expected_include_deferred
+ ):
+ args = cli_parser.get_parser().parse_args(
+ ["pools", "update", "--pool", "p", "--slots", "10", *extra_argv]
+ )
+
+ args.func(args, api_client=mock.MagicMock(spec=Client))
+
+ assert mocked_update.call_args.kwargs["pool_body"] == PoolPatchBody(
+ pool="p", slots=10, include_deferred=expected_include_deferred
+ )
+
+ @mock.patch.object(AirflowConsole, "print_as", autospec=True)
+ @mock.patch.object(PoolsOperations, "update", autospec=True)
+ def test_pools_update_requires_include_deferred(self, mocked_update,
mocked_print_as):
+ """Omitting ``--include-deferred`` must neither send
``include_deferred=False`` nor reach the server."""
+ args = cli_parser.get_parser().parse_args(["pools", "update",
"--pool", "p", "--slots", "10"])
+
+ with pytest.raises(SystemExit, match="Provide --include-deferred or
--no-include-deferred"):
+ args.func(args, api_client=mock.MagicMock(spec=Client))
+
+ mocked_update.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("extra_argv", "expected_run_on_latest_version"),
+ [
+ pytest.param([], None, id="omitted-flag-stays-unset"),
+ pytest.param(["--run-on-latest-version"], True,
id="explicit-run-on-latest"),
+ pytest.param(["--no-run-on-latest-version"], False,
id="explicit-run-on-original"),
+ ],
+ )
+ @mock.patch.object(AirflowConsole, "print_as", autospec=True)
+ @mock.patch.object(BackfillOperations, "create", autospec=True)
+ def test_backfill_create_omitted_run_on_latest_version_stays_unset(
+ self, mocked_create, mocked_print_as, extra_argv,
expected_run_on_latest_version
+ ):
+ """Omitting ``--run-on-latest-version`` must not send
``run_on_latest_version=False``."""
+ args = cli_parser.get_parser().parse_args(
+ [
+ "backfill",
+ "create",
+ "--dag-id",
+ "d",
+ "--from-date",
+ "2024-01-01T00:00:00Z",
+ "--to-date",
+ "2024-01-02T00:00:00Z",
+ *extra_argv,
+ ]
+ )
+
+ args.func(args, api_client=mock.MagicMock(spec=Client))
+
+ backfill = mocked_create.call_args.kwargs["backfill"]
+ assert isinstance(backfill, BackfillPostBody)
+ assert backfill.run_on_latest_version is expected_run_on_latest_version
+
def test_parser_resolves_nested_subcommand(self):
"""The parser descends into the state-store subgroup, binds the func,
and keeps the auth token."""
args = cli_parser.get_parser().parse_args(
