Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 32 additions & 5 deletions airflow-ctl/src/airflowctl/ctl/cli_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

import argparse
import ast
import datetime
import getpass
import inspect
import os
Expand Down Expand Up @@ -441,6 +442,32 @@ def _is_primitive_type(type_name: str) -> bool:
}
return type_name in primitive_types

@staticmethod
def _python_type_from_string(type_name: str) -> type:
"""Return the corresponding Python *type* for a primitive type name string.

This helper is used when generating ``argparse`` CLI arguments from the
OpenAPI-derived operation signatures. Without this mapping the CLI would
incorrectly assume every primitive parameter is a *string*, potentially
leading to type errors or unexpected behaviour when invoking the REST
API.
"""
mapping: dict[str, type] = {
"int": int,
"float": float,
"bool": bool,
"str": str,
"bytes": bytes,
"list": list,
"dict": dict,
"tuple": tuple,
"set": set,
"datetime.datetime": datetime.datetime,
}
# Default to ``str`` to preserve previous behaviour for any unrecognised
# type names while still allowing the CLI to function.
return mapping.get(type_name, str)

@staticmethod
def _create_arg(
arg_flags: tuple,
Expand Down Expand Up @@ -507,15 +534,15 @@ def _create_args_map_from_operation(self):
for parameter in operation.get("parameters"):
for parameter_key, parameter_type in parameter.items():
if self._is_primitive_type(type_name=parameter_type):
python_type = self._python_type_from_string(parameter_type)
is_bool = parameter_type == "bool"
args.append(
self._create_arg(
arg_flags=("--" + self._sanitize_arg_parameter_key(parameter_key),),
arg_type=type(parameter_type),
arg_action=argparse.BooleanOptionalAction
if type(parameter_type) is bool
else None,
arg_type=None if is_bool else python_type,
arg_action=argparse.BooleanOptionalAction if is_bool else None, # type: ignore
arg_help=f"{parameter_key} for {operation.get('name')} operation in {operation.get('parent').name}",
arg_default=False if type(parameter_type) is bool else None,
arg_default=False if is_bool else None,
)
)
else:
Expand Down