in cli/src/pcluster/lib/lib.py [0:0]
def _gen_func_map(model: Dict) -> Dict[str, Callable]: # noqa: C901
"""Generate a dict mapping function names to dispatch functions."""
class Args:
"""An Args class with structure to be used during dispatch."""
def __init__(self, args):
self.__dict__ = args
# Much like the CLI, we add wait commands for these cluster options to
# allow the user to have these be synchronous. This is done at the argument
# parser level on the CLI side.
wait_ops = ["create-cluster", "delete-cluster", "update-cluster"]
for op_name in filter(lambda x: x in model, wait_ops):
wait_param = {"body": False, "type": "boolean", "name": "wait", "required": False}
model[op_name]["params"].append(wait_param)
def make_func(op_name: str) -> Callable:
"""Take the name of an operation and return the function that call the controller."""
def func(**kwargs):
# Validate that args provided match the model
params = model[op_name]["params"]
expected = {to_snake_case(param["name"]) for param in params if param["required"]}
missing = expected - set(kwargs.keys())
if missing:
raise TypeError(f"<{op_name}> missing required arguments: {missing}")
all_args = {to_snake_case(param["name"]) for param in params}
unexpected = set(kwargs.keys()) - all_args - {"query", "debug"}
if unexpected:
raise TypeError(f"<{op_name}> got unexpected arguments: {unexpected}")
kwargs["func"] = None
kwargs["operation"] = op_name
for param in model[op_name]["params"]:
param_name = to_snake_case(param["name"])
if param_name not in kwargs:
kwargs[param_name] = None
# Convert python data to strings for args of type "file"
elif not isinstance(kwargs.get(param_name), str) and param["type"] == "file":
kwargs[param_name] = yaml.dump(kwargs[param_name])
elif isinstance(kwargs.get(param_name), list) and all(
isinstance(item, str) for item in kwargs.get(param_name)
):
pass
else:
kwargs[param_name] = param_coerce(param)(kwargs[param_name])
return dispatch(model, Args(kwargs))
return func
return {to_snake_case(op): make_func(op) for op in model}