in src/azure-cli-core/azure/cli/core/commands/arm.py [0:0]
def register_ids_argument(cli_ctx):
from knack import events
ids_metadata = {}
def add_ids_arguments(_, **kwargs): # pylint: disable=unused-argument
command_table = kwargs.get('commands_loader').command_table
if not command_table:
return
for command in command_table.values():
# Somewhat blunt hammer, but any create commands will not have an automatic id parameter
if command.name.split()[-1] == 'create':
continue
# Only commands with a resource name are candidates for an id parameter
id_parts = [a.type.settings.get('id_part') for a in command.arguments.values()]
if 'name' not in id_parts and 'resource_name' not in id_parts:
continue
group_name = 'Resource Id'
# determine which arguments are required and optional and store in ids_metadata
ids_metadata[command.name] = {'required': [], 'optional': []}
for arg in [a for a in command.arguments.values() if a.type.settings.get('id_part')]:
if arg.options.get('required', False):
ids_metadata[command.name]['required'].append(arg.name)
else:
ids_metadata[command.name]['optional'].append(arg.name)
arg.required = False
arg.arg_group = group_name
# retrieve existing `ids` arg if it exists
id_arg = command.loader.argument_registry.arguments[command.name].get('ids', None)
deprecate_info = id_arg.settings.get('deprecate_info', None) if id_arg else None
id_kwargs = {
'metavar': 'ID',
'help': "One or more resource IDs (space-delimited). "
"It should be a complete resource ID containing all information of '{gname}' arguments. "
"You should provide either --ids or other '{gname}' arguments.".format(gname=group_name),
'dest': 'ids' if id_arg else '_ids',
'deprecate_info': deprecate_info,
'is_preview': id_arg.settings.get('is_preview', None) if id_arg else None,
'is_experimental': id_arg.settings.get('is_experimental', None) if id_arg else None,
'nargs': '+',
'arg_group': group_name
}
command.add_argument('ids', '--ids', **id_kwargs)
def parse_ids_arguments(_, command, args):
namespace = args
cmd = namespace._cmd # pylint: disable=protected-access
# some commands have custom IDs and parsing. This will not work for that.
if not ids_metadata.get(command, None):
return
ids = getattr(namespace, 'ids', getattr(namespace, '_ids', None))
required_args = [cmd.arguments[x] for x in ids_metadata[command]['required']]
optional_args = [cmd.arguments[x] for x in ids_metadata[command]['optional']]
combined_args = required_args + optional_args
if not ids:
# ensure the required parameters are provided if --ids is not
errors = [arg for arg in required_args if getattr(namespace, arg.name, None) is None]
if errors:
missing_required = ' '.join(arg.options_list[0] for arg in errors)
raise CLIError('({} | {}) are required'.format(missing_required, '--ids'))
return
# show warning if names are used in conjunction with --ids
other_values = {arg.name: {'arg': arg, 'value': getattr(namespace, arg.name, None)}
for arg in combined_args}
for _, data in other_values.items():
if data['value'] and not getattr(data['value'], 'is_default', None):
logger.warning("option '%s' will be ignored due to use of '--ids'.",
data['arg'].type.settings['options_list'][0])
# create the empty lists, overwriting any values that may already be there
for arg in combined_args:
setattr(namespace, arg.name, IterateValue())
def assemble_json(ids):
lcount = 0
lind = None
for i, line in enumerate(ids):
if line == '[':
if lcount == 0:
lind = i
lcount += 1
elif line == ']':
lcount -= 1
# final closed set of matching brackets
if lcount == 0:
left = lind
right = i + 1
l_comp = ids[:left]
m_comp = [''.join(ids[left:right])]
r_comp = ids[right:]
ids = l_comp + m_comp + r_comp
return assemble_json(ids)
# base case--no more merging required
return ids
# reassemble JSON strings from bash
ids = assemble_json(ids)
# expand the IDs into the relevant fields
full_id_list = []
for val in ids:
try:
# support piping values from JSON. Does not require use of --query
json_vals = json.loads(val)
if not isinstance(json_vals, list):
json_vals = [json_vals]
for json_val in json_vals:
if isinstance(json_val, dict) and 'id' in json_val:
full_id_list += [json_val['id']]
except ValueError:
# supports piping of --ids to the command when using TSV. Requires use of --query
full_id_list = full_id_list + val.splitlines()
if full_id_list:
setattr(namespace, '_ids', full_id_list)
from azure.mgmt.core.tools import parse_resource_id, is_valid_resource_id
for val in full_id_list:
if not is_valid_resource_id(val):
raise CLIError('invalid resource ID: {}'.format(val))
# place the ID parts into the correct property lists
parts = parse_resource_id(val)
for arg in combined_args:
id_part = arg.type.settings.get('id_part')
id_value = parts.get(id_part, None)
if id_value is None:
argument_name = arg.type.settings.get('options_list')[0]
raise CLIError("Argument {arg_name} cannot be derived from ID {id}. "
"Please provide a complete resource ID "
"containing all information of '{group_name}' "
"arguments. ".format(id=val,
arg_name=argument_name,
group_name=arg.arg_group))
getattr(namespace, arg.name).append(id_value)
# support deprecating --ids
deprecate_info = cmd.arguments['ids'].type.settings.get('deprecate_info')
if deprecate_info:
if not hasattr(namespace, '_argument_deprecations'):
setattr(namespace, '_argument_deprecations', [deprecate_info])
else:
namespace._argument_deprecations.append(deprecate_info) # pylint: disable=protected-access
cli_ctx.register_event(events.EVENT_INVOKER_POST_CMD_TBL_CREATE, add_ids_arguments)
cli_ctx.register_event(events.EVENT_INVOKER_POST_PARSE_ARGS, parse_ids_arguments)