Repository: apache/airflow
File: airflow-ctl/src/airflowctl/ctl/cli_config.py (Python)

Code before:
            datamodel_param_name = None
            args_dict = vars(args)
            for parameter in api_operation["parameters"]:
                for parameter_key, parameter_type in parameter.items():
                    if self._is_primitive_type(type_name=parameter_type):
                        method_params[self._sanitize_method_param_key(parameter_key)] = args_dict[
                            parameter_key
                        ]
                    else:
                        datamodel = getattr(generated_datamodels, parameter_type)
                        for expanded_parameter in self.datamodels_extended_map[parameter_type]:
                            if parameter_key not in method_params:
                                method_params[parameter_key] = {}
                                datamodel_param_name = parameter_key
                            if expanded_parameter in self.excluded_parameters:
                                continue
                            if expanded_parameter in args_dict.keys():
                                method_params[parameter_key][
                                    self._sanitize_method_param_key(expanded_parameter)
                                ] = args_dict[expanded_parameter]

            if datamodel:
                if datamodel_param_name:
                    method_params[datamodel_param_name] = datamodel.model_validate(
                        method_params[datamodel_param_name]
                    )
                else:
                    method_params = datamodel.model_validate(method_params)
                method_output = operation_method_object(**method_params)
            else:
                method_output = operation_method_object(**method_params)

            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:
                """Check if the object is a nested dictionary and collect list of dictionaries."""

                def is_dict_nested(obj: dict) -> bool:
                    """Check if the object is a nested dictionary."""
                    return any(isinstance(i, dict) or isinstance(i, list) for i in obj.values())

                if is_dict_nested(dict_obj):
                    iteration_dict = dict_obj.copy()
                    for key, value in iteration_dict.items():

Review comment (performance):
Small nit, maybe would be great to move this into a method to have clear updates on which datamodels from a single place rather than this method, which already has a lot of cyclomatic complexity :) 

Code after:
            datamodel_param_name = None
            args_dict = vars(args)
            for parameter in api_operation["parameters"]:
                for parameter_key, parameter_type in parameter.items():
                    if self._is_primitive_type(type_name=parameter_type):
                        method_params[self._sanitize_method_param_key(parameter_key)] = args_dict[
                            parameter_key
                        ]
                    else:
                        datamodel = getattr(generated_datamodels, parameter_type)
                        for expanded_parameter in self.datamodels_extended_map[parameter_type]:
                            if parameter_key not in method_params:
                                method_params[parameter_key] = {}
                                datamodel_param_name = parameter_key
                            if expanded_parameter in self.excluded_parameters:
                                continue
                            if expanded_parameter in args_dict.keys():
                                method_params[parameter_key][
                                    self._sanitize_method_param_key(expanded_parameter)
                                ] = args_dict[expanded_parameter]

            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
                        )
                    method_params[datamodel_param_name] = datamodel.model_validate(
                        method_params[datamodel_param_name]
                    )
                else:
                    method_params = datamodel.model_validate(method_params)
                method_output = operation_method_object(**method_params)
            else:
                method_output = operation_method_object(**method_params)

            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