Fix airflowctl operations create command validation error

airflow-ctl/src/airflowctl/ctl/cli_config.py

@@ -27,6 +27,7 @@
 import json
 import os
 import sys
+import typing
 from argparse import Namespace
 from collections.abc import Callable, Iterable
 from enum import Enum
@@ -713,6 +714,7 @@ def _create_arg(
         arg_action: argparse.BooleanOptionalAction | None,
         arg_dest: str | None = None,
         arg_default: Any | None = None,
+        arg_required: bool = False,
     ) -> Arg:
         return Arg(
             flags=arg_flags,
@@ -721,6 +723,7 @@ def _create_arg(
             help=arg_help,
             default=arg_default,
             action=arg_action,
+            required=True if arg_required else _UNSET,
         )
 
     @staticmethod
@@ -743,6 +746,11 @@ def _create_positional_arg(
             help=arg_help,
         )
 
+    @staticmethod
+    def _allows_none(annotation: Any) -> bool:
+        """Return True if the annotation accepts None, e.g. ``str | None``."""
+        return type(None) in typing.get_args(annotation)
+
     def _find_subgroup_suffix(self, operation_group_name: str, operation_name: str) -> str | None:
         """Find the subgroup suffix an operation belongs to, if one exists."""
         for suffix in self.subgroups.get(operation_group_name, set()):
@@ -786,9 +794,16 @@ def _create_arg_for_non_primitive_type(
                     arg_type=self._python_type_from_string(annotation),
                     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,
+                    arg_default=(
+                        self._get_bool_arg_default(parameter_type, field_type.default)
+                        if annotation is bool
+                        else None
+                    ),
+                    arg_required=(
+                        field_type.is_required()  # no default?
+                        and annotation is not bool  # not a bool flag
+                        and not self._allows_none(field_type.annotation)  # None not acceptable
+                    ),
                 )
             )
         return commands

airflow-ctl/tests/airflow_ctl/api/test_operations.py

@@ -935,7 +935,9 @@ def handle_request(request: httpx.Request) -> httpx.Response:
             )
 
         client = make_api_client(transport=httpx.MockTransport(handle_request))
-        response = client.connections.create(connection=self.connection)
+        response = client.connections.create(
+            connection=self.connection,
+        )
         assert response == self.connection_response
 
     def test_create_uses_schema_alias_in_request_body(self):
@@ -959,7 +961,9 @@ def handle_request(request: httpx.Request) -> httpx.Response:
             )
 
         client = make_api_client(transport=httpx.MockTransport(handle_request))
-        response = client.connections.create(connection=connection)
+        response = client.connections.create(
+            connection=connection,
+        )
         assert response == self.connection_response
 
     def test_bulk(self):

airflow-ctl/tests/airflow_ctl/ctl/test_cli_config.py

@@ -574,6 +574,47 @@ def list(self, limit: int = 100) -> WidgetCollectionResponse | ServerResponseErr
         assert limit_arg.flags == ("--limit",)
         assert limit_arg.kwargs["type"] is int
 
+    @pytest.mark.parametrize(
+        ("group", "command", "required_flags"),
+        [
+            pytest.param(
+                "connections", "create", {"--connection-id", "--conn-type"}, id="connections-create"
+            ),
+            pytest.param("pools", "create", {"--name", "--slots"}, id="pools-create"),
+        ],
+    )
+    def test_required_datamodel_fields_become_required_flags(self, group, command, required_flags):
+        """Required fields of a body model are generated as required ``--flags``."""
+        command_factory = CommandFactory()
+        group_command = next(g for g in command_factory.group_commands if g.name == group)
+        sub_command = next(c for c in group_command.subcommands if c.name == command)
+
+        required = {arg.flags[0] for arg in sub_command.args if arg.kwargs.get("required") is True}
+        assert required == required_flags
+
+        for arg in sub_command.args:
+            action = arg.kwargs.get("action")
+            if action is BooleanOptionalAction:
+                assert not arg.kwargs.get("required"), "bool flags always have a default"
+
+    def test_optional_datamodel_fields_default_to_none(self):
+        """Optional non-bool fields stay ``None`` so Pydantic sees them as not provided."""
+        command_factory = CommandFactory()
+        group_command = next(g for g in command_factory.group_commands if g.name == "connections")
+        sub_command = next(c for c in group_command.subcommands if c.name == "create")
+
+        parser = argparse.ArgumentParser()
+        for arg in sub_command.args:
+            arg.add_to_parser(parser)
+
+        namespace = parser.parse_args(["--connection-id", "c1", "--conn-type", "http"])
+
+        assert namespace.connection_id == "c1"
+        assert namespace.conn_type == "http"
+        assert namespace.host is None
+        assert namespace.login is None
+        assert namespace.description is None
+
     def test_every_generated_command_accepts_the_output_flag(self):
         """``_get_func`` always prints through ``args.output``, so every generated command must declare it."""
         command_factory = CommandFactory()
@@ -598,6 +639,15 @@ def iter_action_commands(commands, prefix):
 
         assert missing == []
 
+    def test_required_but_nullable_datamodel_field_stays_optional(self):
+        """A field with no default that accepts None (e.g. ``logical_date``) must not become a required flag."""
+        command_factory = CommandFactory()
+        dags_group = next(g for g in command_factory.group_commands if g.name == "dags")
+        trigger = next(c for c in dags_group.subcommands if c.name == "trigger")
+
+        logical_date_arg = next(arg for arg in trigger.args if arg.flags == ("--logical-date",))
+        assert not logical_date_arg.kwargs.get("required")
+
 
 class TestCliConfigMethods:
     @pytest.mark.parametrize(