Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_arguments_loader(context, getter_op, cmd_args=None, operation_group=None):
getter_args = dict(extract_args_from_signature(context.get_op_handler(getter_op, operation_group=operation_group),
excluded_params=EXCLUDED_PARAMS))
cmd_args = cmd_args or {}
cmd_args.update(getter_args)
cmd_args['cmd'] = CLICommandArgument('cmd', arg_type=ignore_type)
return cmd_args
def expand(self, dest, model_type, group_name=None, patches=None):
super(RdbmsArgumentContext, self).expand(dest, model_type, group_name, patches)
from knack.arguments import ignore_type
# Remove the validator and store it into a list
arg = self.command_loader.argument_registry.arguments[self.command_scope].get(dest, None)
if not arg: # when the argument context scope is N/A
return
self.validators.append(arg.settings['validator'])
dest_option = ['--__{}'.format(dest.upper())]
if dest == 'parameters':
from .validators import get_combined_validator
self.argument(dest,
arg_type=ignore_type,
options_list=dest_option,
validator=get_combined_validator(self.validators))
else:
self.argument(dest, options_list=dest_option, arg_type=ignore_type, validator=None)
# Remove the validator and store it into a list
arg = self.command_loader.argument_registry.arguments[self.command_scope].get(dest, None)
if not arg: # when the argument context scope is N/A
return
self.validators.append(arg.settings['validator'])
dest_option = ['--__{}'.format(dest.upper())]
if dest == 'parameters':
from .validators import get_combined_validator
self.argument(dest,
arg_type=ignore_type,
options_list=dest_option,
validator=get_combined_validator(self.validators))
else:
self.argument(dest, options_list=dest_option, arg_type=ignore_type, validator=None)
c.argument('max_connections', arg_group='Download Control', type=int)
c.argument('validate_content', action='store_true', min_api='2016-05-31')
c.extra('no_progress', progress_type)
with self.argument_context('storage file delete-batch') as c:
from ._validators import process_file_batch_source_parameters
c.argument('source', options_list=('--source', '-s'), validator=process_file_batch_source_parameters)
with self.argument_context('storage file copy start') as c:
from azure.cli.command_modules.storage._validators import validate_source_uri
c.register_source_uri_arguments(validator=validate_source_uri)
with self.argument_context('storage file copy start-batch', arg_group='Copy Source') as c:
from ._validators import get_source_file_or_blob_service_client
c.argument('source_client', ignore_type, validator=get_source_file_or_blob_service_client)
c.extra('source_account_name')
c.extra('source_account_key')
c.extra('source_uri')
c.argument('source_sas')
c.argument('source_container')
c.argument('source_share')
with self.argument_context('storage cors list') as c:
c.extra('services', validator=get_char_options_validator('bfqt', 'services'), default='bqft',
options_list='--services', required=False)
with self.argument_context('storage cors add') as c:
c.extra('services', validator=get_char_options_validator('bfqt', 'services'), required=True,
options_list='--services')
c.argument('max_age')
c.argument('origins', nargs='+')
command_loaders = set()
for loaders in self.cmd_to_loader_map.values():
command_loaders = command_loaders.union(set(loaders))
logger.info('Applying %s command loaders...', len(command_loaders))
else:
command_loaders = self.cmd_to_loader_map.get(command, None)
if command_loaders:
for loader in command_loaders:
# register global args
with loader.argument_context('') as c:
c.argument('resource_group_name', resource_group_name_type)
c.argument('location', get_location_type(self.cli_ctx))
c.argument('deployment_name', deployment_name_type)
c.argument('cmd', ignore_type)
if command is None:
# load all arguments via reflection
for cmd in loader.command_table.values():
cmd.load_arguments() # this loads the arguments via reflection
loader.skip_applicability = True
loader.load_arguments('') # this adds entries to the argument registries
else:
loader.command_name = command
self.command_table[command].load_arguments() # this loads the arguments via reflection
loader.load_arguments(command) # this adds entries to the argument registries
self.argument_registry.arguments.update(loader.argument_registry.arguments)
self.extra_argument_registry.update(loader.extra_argument_registry)
loader._update_command_definitions() # pylint: disable=protected-access
def _ignore_if_not_registered(self, dest):
scope = self.scope
arg_registry = self.command_loader.argument_registry
match = arg_registry.arguments[scope].get(dest, {})
if not match:
super(AzArgumentContext, self).argument(dest, arg_type=ignore_type)
c.argument('tags', tags_type)
c.argument('relative_record_set_name', name_arg_type, help='The name of the record set, relative to the name of the Private DNS zone.')
c.argument('private_zone_name', options_list=('--zone-name', '-z'), help='The name of the Private DNS zone.', type=privatedns_zone_name_type)
c.argument('metadata', tags_type, help='Metadata in space-separated key=value pairs. This overwrites any existing metadata.', validator=validate_privatedns_metadata)
with self.argument_context('network private-dns zone') as c:
c.argument('private_zone_name', name_arg_type, type=privatedns_zone_name_type)
c.ignore('location')
with self.argument_context('network private-dns link') as c:
c.argument('virtual_network_link_name', name_arg_type, help='The name of the virtual network link to the specified Private DNS zone.')
c.argument('virtual_network', help='Name or ID of the virtual network.', options_list=('--virtual-network', '-v'), validator=get_vnet_validator)
c.argument('registration_enabled', help='Specify if the link is registration enabled.', options_list=('--registration-enabled', '-e'), arg_type=get_three_state_flag())
with self.argument_context('network private-dns record-set') as c:
c.argument('record_type', ignore_type, validator=validate_privatedns_record_type)
for item in ['', 'a', 'aaaa', 'cname', 'mx', 'ptr', 'srv', 'txt']:
with self.argument_context('network private-dns record-set {} create'.format(item)) as c:
c.argument('ttl', help='Record set TTL (time-to-live)')
for item in ['a', 'aaaa', 'cname', 'mx', 'ptr', 'srv', 'txt']:
with self.argument_context('network private-dns record-set {} add-record'.format(item)) as c:
c.argument('relative_record_set_name', options_list=('--record-set-name', '-n'), help='The name of the record set relative to the zone. Creates a new record set if one does not exist.')
with self.argument_context('network private-dns record-set {} remove-record'.format(item)) as c:
c.argument('relative_record_set_name', options_list=('--record-set-name', '-n'), help='The name of the record set relative to the zone.')
c.argument('keep_empty_record_set', action='store_true', help='Keep the empty record set if the last record is removed.')
with self.argument_context('network private-dns record-set cname set-record') as c:
c.argument('relative_record_set_name', options_list=['--record-set-name', '-n'], help='The name of the record set relative to the zone. Creates a new record set if one does not exist.')
options_list=['--admin-user', '-u'])
c.argument('administrator_login_password',
options_list=['--admin-password', '-p'])
c.argument('authentication_type',
options_list=['--auth-type', '-a'],
arg_type=get_enum_type(AuthenticationType))
c.argument('storage_key_type',
arg_type=get_enum_type(StorageKeyType))
# The parameter name '--name' is used for 'database_name', so we need to give a different name
# for the import extension 'name' parameter to avoid conflicts. This parameter is actually not
# needed, but we still need to avoid this conflict.
c.argument('name', options_list=['--not-name'], arg_type=ignore_type)
with self.argument_context('sql db show-connection-string') as c:
c.argument('client_provider',
options_list=['--client', '-c'],
help='Type of client connection provider.',
arg_type=get_enum_type(ClientType))
auth_group = 'Authentication'
c.argument('auth_type',
options_list=['--auth-type', '-a'],
arg_group=auth_group,
help='Type of authentication.',
arg_type=get_enum_type(ClientAuthenticationType))
#####
with self.argument_context('network private-dns record-set {} create'.format(item)) as c:
c.argument('ttl', help='Record set TTL (time-to-live)')
for item in ['a', 'aaaa', 'cname', 'mx', 'ptr', 'srv', 'txt']:
with self.argument_context('network private-dns record-set {} add-record'.format(item)) as c:
c.argument('relative_record_set_name', options_list=('--record-set-name', '-n'), help='The name of the record set relative to the zone. Creates a new record set if one does not exist.')
with self.argument_context('network private-dns record-set {} remove-record'.format(item)) as c:
c.argument('relative_record_set_name', options_list=('--record-set-name', '-n'), help='The name of the record set relative to the zone.')
c.argument('keep_empty_record_set', action='store_true', help='Keep the empty record set if the last record is removed.')
with self.argument_context('network private-dns record-set cname set-record') as c:
c.argument('relative_record_set_name', options_list=['--record-set-name', '-n'], help='The name of the record set relative to the zone. Creates a new record set if one does not exist.')
with self.argument_context('network private-dns record-set soa') as c:
c.argument('relative_record_set_name', ignore_type, default='@')
with self.argument_context('network private-dns record-set a') as c:
c.argument('ipv4_address', options_list=('--ipv4-address', '-a'), help='IPV4 address in string notation.')
with self.argument_context('network private-dns record-set aaaa') as c:
c.argument('ipv6_address', options_list=('--ipv6-address', '-a'), help='IPV6 address in string notation.')
with self.argument_context('network private-dns record-set cname') as c:
c.argument('cname', options_list=('--cname', '-c'), help='Canonical name.')
with self.argument_context('network private-dns record-set mx') as c:
c.argument('exchange', options_list=('--exchange', '-e'), help='Exchange metric.')
c.argument('preference', options_list=('--preference', '-p'), help='Preference metric.')
with self.argument_context('network private-dns record-set ptr') as c:
c.argument('dname', options_list=('--ptrdname', '-d'), help='PTR target domain name.')