-
Notifications
You must be signed in to change notification settings - Fork 16.3k
Open
Assignees
Labels
Description
Description
What
We need to make adjustments in this file.
airflow/airflow-ctl/src/airflowctl/ctl/cli_config.py
Lines 499 to 584 in a4f3417
| def _create_arg( | |
| arg_flags: tuple, | |
| arg_type: type | Callable, | |
| arg_help: str, | |
| arg_action: argparse.BooleanOptionalAction | None, | |
| arg_dest: str | None = None, | |
| arg_default: Any | None = None, | |
| ) -> Arg: | |
| return Arg( | |
| flags=arg_flags, | |
| type=arg_type, | |
| dest=arg_dest, | |
| help=arg_help, | |
| default=arg_default, | |
| action=arg_action, | |
| ) | |
| def _create_arg_for_non_primitive_type( | |
| self, | |
| parameter_type: str, | |
| parameter_key: str, | |
| ) -> list[Arg]: | |
| """Create Arg for non-primitive type Pydantic.""" | |
| parameter_type_map = getattr(generated_datamodels, parameter_type) | |
| commands = [] | |
| if parameter_type_map not in self.datamodels_extended_map.keys(): | |
| self.datamodels_extended_map[parameter_type] = [] | |
| for field, field_type in parameter_type_map.model_fields.items(): | |
| if field in self.excluded_parameters: | |
| continue | |
| self.datamodels_extended_map[parameter_type].append(field) | |
| if type(field_type.annotation) is type: | |
| commands.append( | |
| self._create_arg( | |
| arg_flags=("--" + self._sanitize_arg_parameter_key(field),), | |
| arg_type=self._python_type_from_string(field_type.annotation), | |
| arg_action=argparse.BooleanOptionalAction if field_type.annotation is bool else None, # type: ignore | |
| arg_help=f"{field} for {parameter_key} operation", | |
| arg_default=False if field_type.annotation is bool else None, | |
| ) | |
| ) | |
| else: | |
| try: | |
| annotation = field_type.annotation.__args__[0] | |
| except AttributeError: | |
| annotation = field_type.annotation | |
| commands.append( | |
| self._create_arg( | |
| arg_flags=("--" + self._sanitize_arg_parameter_key(field),), | |
| 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=False if annotation is bool else None, | |
| ) | |
| ) | |
| return commands | |
| def _create_args_map_from_operation(self): | |
| """Create Arg from Operation Method checking for parameters and return types.""" | |
| for operation in self.operations: | |
| args = [] | |
| for parameter in operation.get("parameters"): | |
| for parameter_key, parameter_type in parameter.items(): | |
| if self._is_primitive_type(type_name=parameter_type): | |
| is_bool = parameter_type == "bool" | |
| args.append( | |
| self._create_arg( | |
| arg_flags=("--" + self._sanitize_arg_parameter_key(parameter_key),), | |
| arg_type=self._python_type_from_string(parameter_type), | |
| arg_action=argparse.BooleanOptionalAction if is_bool else None, | |
| arg_help=f"{parameter_key} for {operation.get('name')} operation in {operation.get('parent').name}", | |
| arg_default=False if is_bool else None, | |
| ) | |
| ) | |
| else: | |
| args.extend( | |
| self._create_arg_for_non_primitive_type( | |
| parameter_type=parameter_type, parameter_key=parameter_key | |
| ) | |
| ) | |
| if any(operation.get("name").startswith(cmd) for cmd in self.output_command_list): | |
| args.extend([ARG_OUTPUT, ARG_AUTH_ENVIRONMENT]) | |
| self.args_map[(operation.get("name"), operation.get("parent").name)] = args |
airflowctl will understand whether the parameter is nullable. If it is not nullable, we should add as a destination parameter, and if it is nullable, it will be positional.
Example:
from
airflow connections create --connection-id="test" --conn-type="test" ...
to
airflow connections create test test ...
Are you willing to submit a PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct
Prab-27