Refactor datamodel defaulting logic into dedicated method (#61236)

Extract TriggerDAGRunPostBody logical_date defaulting from _get_func into
  a new _apply_datamodel_defaults method. This addresses review feedback to
  reduce cyclomatic complexity in the already-complex _get_func method and
  provides a centralized location for managing datamodel-specific defaults.

Author: dheerajturaga

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

@@ -586,6 +586,31 @@ def _create_args_map_from_operation(self):
 
             self.args_map[(operation.get("name"), operation.get("parent").name)] = args
 
+    def _apply_datamodel_defaults(self, datamodel: type, params: dict) -> dict:
+        """
+        Apply datamodel-specific default values.
+
+        Centralizes special handling for different datamodels to keep
+        CLI parameter processing logic clean and maintainable.
+
+        Args:
+            datamodel: The Pydantic datamodel class
+            params: Dictionary of parameters for the datamodel
+
+        Returns:
+            Updated params dictionary with defaults applied
+        """
+        # Handle TriggerDAGRunPostBody: default logical_date to now
+        # This matches the Airflow UI behavior where the form pre-fills with current time
+        if (
+            datamodel.__name__ == "TriggerDAGRunPostBody"
+            and "logical_date" in params
+            and params["logical_date"] is None
+        ):
+            params["logical_date"] = datetime.datetime.now(datetime.timezone.utc)
+
+        return params
+
     def _create_func_map_from_operation(self):
         """Create function map from Operation Method checking for parameters and return types."""
 
@@ -624,16 +649,10 @@ def _get_func(args: Namespace, api_operation: dict, api_client: Client = NEW_API
 
             if datamodel:
                 if datamodel_param_name:
-                    # Special handling for TriggerDAGRunPostBody: default logical_date to now
-                    # This matches the Airflow UI behavior where the form pre-fills with current time
-                    if (
-                        datamodel.__name__ == "TriggerDAGRunPostBody"
-                        and "logical_date" in method_params[datamodel_param_name]
-                        and method_params[datamodel_param_name]["logical_date"] is None
-                    ):
-                        method_params[datamodel_param_name]["logical_date"] = datetime.datetime.now(
-                            datetime.timezone.utc
-                        )
+                    # Apply datamodel-specific defaults (e.g., logical_date for TriggerDAGRunPostBody)
+                    method_params[datamodel_param_name] = self._apply_datamodel_defaults(
+                        datamodel, method_params[datamodel_param_name]
+                    )
                     method_params[datamodel_param_name] = datamodel.model_validate(
                         method_params[datamodel_param_name]
                     )

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

@@ -403,3 +403,67 @@ def test_trigger_dag_run_defaults_logical_date_to_now(self):
 
         # Also verify timezone is UTC
         assert trigger_body.logical_date.tzinfo is not None, "logical_date should have timezone info"
+
+    def test_apply_datamodel_defaults_trigger_dag_run_with_none(self):
+        """Test _apply_datamodel_defaults sets logical_date to now when None for TriggerDAGRunPostBody."""
+        from datetime import datetime, timezone
+
+        from airflowctl.api.datamodels.generated import TriggerDAGRunPostBody
+
+        command_factory = CommandFactory()
+
+        # Test with logical_date=None
+        params = {"logical_date": None, "conf": {}}
+        result = command_factory._apply_datamodel_defaults(TriggerDAGRunPostBody, params)
+
+        assert result["logical_date"] is not None, "logical_date should be defaulted to now"
+        assert isinstance(result["logical_date"], datetime)
+
+        # Verify it's close to current time (within 5 seconds)
+        time_diff = abs((datetime.now(timezone.utc) - result["logical_date"]).total_seconds())
+        assert time_diff < 5, f"logical_date should be close to now, but diff is {time_diff} seconds"
+
+        # Verify timezone is UTC
+        assert result["logical_date"].tzinfo is not None, "logical_date should have timezone info"
+
+    def test_apply_datamodel_defaults_trigger_dag_run_with_value(self):
+        """Test _apply_datamodel_defaults preserves existing logical_date for TriggerDAGRunPostBody."""
+        from datetime import datetime, timezone
+
+        from airflowctl.api.datamodels.generated import TriggerDAGRunPostBody
+
+        command_factory = CommandFactory()
+
+        # Test with an existing logical_date value
+        specific_date = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc)
+        params = {"logical_date": specific_date, "conf": {}}
+        result = command_factory._apply_datamodel_defaults(TriggerDAGRunPostBody, params)
+
+        # Should preserve the provided value, not override it
+        assert result["logical_date"] == specific_date, "logical_date should not be changed when already set"
+
+    def test_apply_datamodel_defaults_trigger_dag_run_without_logical_date(self):
+        """Test _apply_datamodel_defaults doesn't add logical_date if not present."""
+        from airflowctl.api.datamodels.generated import TriggerDAGRunPostBody
+
+        command_factory = CommandFactory()
+
+        # Test without logical_date key
+        params = {"conf": {}}
+        result = command_factory._apply_datamodel_defaults(TriggerDAGRunPostBody, params)
+
+        # Should not add logical_date if it wasn't in params
+        assert "logical_date" not in result, "logical_date should not be added if not originally present"
+
+    def test_apply_datamodel_defaults_other_datamodel(self):
+        """Test _apply_datamodel_defaults doesn't modify params for other datamodels."""
+        from airflowctl.api.datamodels.generated import BackfillPostBody
+
+        command_factory = CommandFactory()
+
+        # Test with a different datamodel (BackfillPostBody)
+        params = {"dag_id": "test_dag", "from_date": None}
+        result = command_factory._apply_datamodel_defaults(BackfillPostBody, params)
+
+        # Should return params unchanged for other datamodels
+        assert result == params, "Params should be unchanged for non-TriggerDAGRunPostBody datamodels"