def list_commands(self, ctx):
self._load_plugin_commands()
# The commands available is the list of both the application (if
# available) plus the builtin commands.
rv = set(click.Group.list_commands(self, ctx))
info = ctx.ensure_object(ScriptInfo)
try:
rv.update(info.load_app().cli.list_commands(ctx))
except Exception:
# Here we intentionally swallow all exceptions as we don't
# want the help page to break if the app does not exist.
# If someone attempts to use the command we try to create
# the app again and this will give us the error.
pass
return sorted(rv)
python类command()的实例源码
def main(as_module=False):
this_module = __package__ + '.cli'
args = sys.argv[1:]
if as_module:
if sys.version_info >= (2, 7):
name = 'python -m ' + this_module.rsplit('.', 1)[0]
else:
name = 'python -m ' + this_module
# This module is always executed as "python -m flask.run" and as such
# we need to ensure that we restore the actual command line so that
# the reloader can properly operate.
sys.argv = ['-m', this_module] + sys.argv[1:]
else:
name = None
cli.main(args=args, prog_name=name)
def list_commands(self, ctx):
self._load_plugin_commands()
# The commands available is the list of both the application (if
# available) plus the builtin commands.
rv = set(click.Group.list_commands(self, ctx))
info = ctx.ensure_object(ScriptInfo)
try:
rv.update(info.load_app().cli.list_commands(ctx))
except Exception:
# Here we intentionally swallow all exceptions as we don't
# want the help page to break if the app does not exist.
# If someone attempts to use the command we try to create
# the app again and this will give us the error.
pass
return sorted(rv)
def main(as_module=False):
this_module = __package__ + '.cli'
args = sys.argv[1:]
if as_module:
if sys.version_info >= (2, 7):
name = 'python -m ' + this_module.rsplit('.', 1)[0]
else:
name = 'python -m ' + this_module
# This module is always executed as "python -m flask.run" and as such
# we need to ensure that we restore the actual command line so that
# the reloader can properly operate.
sys.argv = ['-m', this_module] + sys.argv[1:]
else:
name = None
cli.main(args=args, prog_name=name)
def nsupdate_cmd(name, domain, host_records):
"""
Create nsupdate command line.
"""
pathname = "%s.%s." % (name, domain)
cmd = "server %s\n" % (DNS_SERVER1,)
cmd += "zone %s.\n" % (domain,)
cmd += "update delete %s\n" % (pathname,)
for rec in host_records:
cmd += "update add %s %d %s %s\n" % (pathname, rec[4], rec[2], rec[3])
cmd += "show\n"
cmd += "send\n"
return cmd.encode('utf-8')
def list_commands(self, ctx):
self._load_plugin_commands()
# The commands available is the list of both the application (if
# available) plus the builtin commands.
rv = set(click.Group.list_commands(self, ctx))
info = ctx.ensure_object(ScriptInfo)
try:
rv.update(info.load_app().cli.list_commands(ctx))
except Exception:
# Here we intentionally swallow all exceptions as we don't
# want the help page to break if the app does not exist.
# If someone attempts to use the command we try to create
# the app again and this will give us the error.
pass
return sorted(rv)
def main(as_module=False):
this_module = __package__ + '.cli'
args = sys.argv[1:]
if as_module:
if sys.version_info >= (2, 7):
name = 'python -m ' + this_module.rsplit('.', 1)[0]
else:
name = 'python -m ' + this_module
# This module is always executed as "python -m flask.run" and as such
# we need to ensure that we restore the actual command line so that
# the reloader can properly operate.
sys.argv = ['-m', this_module] + sys.argv[1:]
else:
name = None
cli.main(args=args, prog_name=name)
def test_display_task(self):
@click.command()
@pass_ovh
def cli(ovh):
ovh.display_task({'function': 'foo', 'status': 'init'})
ovh.display_task({'function': 'foo', 'status': 'todo'})
ovh.display_task({'function': 'foo', 'status': 'doing'})
ovh.display_task({'function': 'foo', 'status': 'done'})
ovh.display_task({'function': 'foo', 'status': 'cancelled'})
ovh.display_task({'function': 'foo', 'status': 'bar'})
result = self.runner.invoke(cli)
self.assertEqual(result.output, """\
[*] The task foo has been launched.
[*] The task foo has been launched.
[*] The task foo has been launched.
[*] The task foo is done.
[warning] The task foo has been cancelled.
[error] The task foo fell in an error state.
""")
def list_commands(self, ctx):
self._load_plugin_commands()
# The commands available is the list of both the application (if
# available) plus the builtin commands.
rv = set(click.Group.list_commands(self, ctx))
info = ctx.ensure_object(ScriptInfo)
try:
rv.update(info.load_app().cli.list_commands(ctx))
except Exception:
# Here we intentionally swallow all exceptions as we don't
# want the help page to break if the app does not exist.
# If someone attempts to use the command we try to create
# the app again and this will give us the error.
pass
return sorted(rv)
def main(as_module=False):
this_module = __package__ + '.cli'
args = sys.argv[1:]
if as_module:
if sys.version_info >= (2, 7):
name = 'python -m ' + this_module.rsplit('.', 1)[0]
else:
name = 'python -m ' + this_module
# This module is always executed as "python -m flask.run" and as such
# we need to ensure that we restore the actual command line so that
# the reloader can properly operate.
sys.argv = ['-m', this_module] + sys.argv[1:]
else:
name = None
cli.main(args=args, prog_name=name)
def test_param_with_default__uses_param_default_when_missing(self, cli_runner_isolated):
assert ConfigFileProcessor1.config_files[0] == "hello.ini"
CONFIG_FILE_CONTENTS = """
[hello]
name = Alice
"""
write_configfile_with_contents("hello.ini", CONFIG_FILE_CONTENTS)
assert os.path.exists("hello.ini")
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor1.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", type=str, default="__CMDLINE__")
@click.option("--number", default=123)
def hello_with_param_default(name, number):
click.echo("param: number= %s" % number)
result = cli_runner_isolated.invoke(hello_with_param_default)
assert result.output == "param: number= 42\n"
assert result.exit_code == 0
def test_param_with_default__uses_cmdline_when_provided(self, cli_runner_isolated):
assert ConfigFileProcessor1.config_files[0] == "hello.ini"
CONFIG_FILE_CONTENTS = """
[hello]
name = Alice
number = 43
"""
write_configfile_with_contents("hello.ini", CONFIG_FILE_CONTENTS)
assert os.path.exists("hello.ini")
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor1.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", type=str, default="__CMDLINE__")
@click.option("--number", default=123)
def hello_with_param_default2(name, number):
click.echo("param: number= %s" % number)
result = cli_runner_isolated.invoke(hello_with_param_default2,
["--number", "234"])
assert result.output == "param: number= 234\n"
assert result.exit_code == 0
def test_param_without_default__uses_cmdline_default_when_missing(self, cli_runner_isolated):
assert ConfigFileProcessor1.config_files[0] == "hello.ini"
CONFIG_FILE_CONTENTS = """
[hello]
number = 1234
"""
write_configfile_with_contents("hello.ini", CONFIG_FILE_CONTENTS)
assert os.path.exists("hello.ini")
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor1.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", type=str, default="__CMDLINE__")
def hello_with_config(name):
click.echo("param: name= %s" % name)
result = cli_runner_isolated.invoke(hello_with_config)
assert result.output == "param: name= __CMDLINE__\n"
assert result.exit_code == 0
def test_with_cmdline_and_configfile__prefers_cmdline(self, cli_runner_isolated):
assert ConfigFileProcessor1.config_files[0] == "hello.ini"
CONFIG_FILE_CONTENTS1 = """
[hello]
name = Alice
"""
write_configfile_with_contents("hello.ini", CONFIG_FILE_CONTENTS1)
assert os.path.exists("hello.ini")
assert not os.path.exists("hello.cfg")
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor1.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", default="__CMDLINE__")
def hello(name):
click.echo("Hello %s" % name)
result = cli_runner_isolated.invoke(hello, ["--name", "CMDLINE_VALUE"])
assert result.output == "Hello CMDLINE_VALUE\n"
assert result.exit_code == 0
def test_with_configfile2__usable_as_alternative(self, cli_runner_isolated):
assert ConfigFileProcessor1.config_files[1] == "hello.cfg"
CONFIG_FILE_CONTENTS2 = """
[hello]
name = Bob
"""
write_configfile_with_contents("hello.cfg", CONFIG_FILE_CONTENTS2)
assert not os.path.exists("hello.ini")
assert os.path.exists("hello.cfg")
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor1.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", default="__CMDLINE__")
def hello(name):
click.echo("Hello %s" % name)
result = cli_runner_isolated.invoke(hello)
assert result.output == "Hello Bob\n"
assert result.exit_code == 0
def test_with_configfile12__prefers_configfile1(self, cli_runner_isolated):
assert ConfigFileProcessor1.config_files == ["hello.ini", "hello.cfg"]
CONFIG_FILE_CONTENTS1 = """
[hello]
name = alice
"""
CONFIG_FILE_CONTENTS2 = """
[hello]
name = bob
"""
write_configfile_with_contents("hello.ini", CONFIG_FILE_CONTENTS1)
write_configfile_with_contents("hello.cfg", CONFIG_FILE_CONTENTS2)
assert os.path.exists("hello.ini")
assert os.path.exists("hello.cfg")
CONTEXT_SETTINGS = dict(default_map=ConfigFileProcessor1.read_config())
@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("-n", "--name", default="__CMDLINE__")
def hello(name):
click.echo("Hello %s" % name)
result = cli_runner_isolated.invoke(hello)
assert result.output == "Hello alice\n"
assert result.exit_code == 0
def list_commands(self, ctx):
self._load_plugin_commands()
# The commands available is the list of both the application (if
# available) plus the builtin commands.
rv = set(click.Group.list_commands(self, ctx))
info = ctx.ensure_object(ScriptInfo)
try:
rv.update(info.load_app().cli.list_commands(ctx))
except Exception:
# Here we intentionally swallow all exceptions as we don't
# want the help page to break if the app does not exist.
# If someone attempts to use the command we try to create
# the app again and this will give us the error.
pass
return sorted(rv)
def main(as_module=False):
this_module = __package__ + '.cli'
args = sys.argv[1:]
if as_module:
if sys.version_info >= (2, 7):
name = 'python -m ' + this_module.rsplit('.', 1)[0]
else:
name = 'python -m ' + this_module
# This module is always executed as "python -m flask.run" and as such
# we need to ensure that we restore the actual command line so that
# the reloader can properly operate.
sys.argv = ['-m', this_module] + sys.argv[1:]
else:
name = None
cli.main(args=args, prog_name=name)
def name_to_command(parent, name):
try:
if sys.version_info[0] == 2:
if parent:
parent = parent.encode('ascii', 'replace')
name = name.encode('ascii', 'replace')
if parent:
mod_name = 'tripaille.commands.%s.%s' % (parent, name)
else:
mod_name = 'tripaille.commands.cmd_' + name
mod = __import__(mod_name, None, None, ['cli'])
except ImportError as e:
error("Problem loading command %s, exception %s" % (name, e))
return
return mod.cli
def cli_kick(template_name, extra_vars, limit):
"""
Start an ansible tower job from the command line
"""
try:
# verify configuration
config = Config(config_file())
guard = Guard(config)
template_id = guard.get_template_id(template_name)
extra_v = {}
for extra_var in extra_vars:
extra_v.update(extra_var_to_dict(extra_var))
job = guard.kick(template_id=template_id, limit=limit,
extra_vars=extra_v)
job_url = guard.launch_data_to_url(job)
print('Started job: {0}'.format(job_url))
except CLIError as error:
print(error)
sys.exit(1)
except GuardError as error:
msg = 'Error kicking job tempate: {0} - {1}'.format(template_name,
error)
print(msg)
sys.exit(1)
def __init__(self, main, conf_dir=None, commands_dir=None, **kwargs):
self._commands_dir = commands_dir
if conf_dir:
configure.load_config_file(conf_dir)
# This is a bit of a hack, but need to register all parameters from
# the command line because want to allow tornado to handle them and
# click doesn't contain an equivalent of the `allow_extra_args`
# keyword argument that works for options.
# TODO: don't use tornado command line parsing
params = _options()
super(Command, self).__init__(self,
params=params,
callback=run(main),
invoke_without_command=True,
**kwargs)
def _get_commands(self):
"""Fetch all commands from plugins providing any."""
import collections
result = collections.OrderedDict()
for name, hook in self.hooks.items():
try:
commands = hook(self, pass_octoprint_ctx)
for command in commands:
if not isinstance(command, click.Command):
self._logger.warn("Plugin {} provided invalid CLI command, ignoring it: {!r}".format(name, command))
continue
result[name + self.sep + command.name] = command
except:
self._logger.exception("Error while retrieving cli commants for plugin {}".format(name))
return result
def register(self, command, description, function, params=[], plugin=None):
"""
Registers a new command, which can be used on a command line interface (cli).
:param command: Name of the command
:param description: Description of the command. Is used as help message on cli
:param function: function reference, which gets invoked if command gets called.
:param params: list of click options and arguments
:param plugin: the plugin, which registered this command
:return: command object
"""
if command in self._commands.keys():
raise CommandExistException("Command %s already registered by %s" % (command,
self._commands[command].plugin.name))
new_command = Command(command, description, params, function, plugin)
self._commands[command] = new_command
self._click_root_command.add_command(new_command.click_command)
self.log.debug("Command registered: %s" % command)
return new_command
def unregister(self, command):
"""
Unregisters an existing command, so that this command is no longer available on the command line interface.
This function is mainly used during plugin deactivation.
:param command: Name of the command
"""
if command not in self._commands.keys():
self.log.warning("Can not unregister command %s" % command)
else:
# Click does not have any kind of a function to unregister/remove/deactivate already added commands.
# So we need to delete the related objects manually from the click internal commands dictionary for
# our root command.
del(self._click_root_command.commands[command])
# Finally lets delete the command from our internal dictionary too.
del(self._commands[command])
self.log.debug("Command %s got unregistered" % command)
def _unwrap_func(cls, decorated_func):
'''This unwraps a decorated func, returning the inner wrapped func.
This may become unnecessary with Python 3.4's inspect.unwrap().
'''
if click is not None:
# Workaround for click.command() decorator not setting
# __wrapped__
if isinstance(decorated_func, click.Command):
return cls._unwrap_func(decorated_func.callback)
if hasattr(decorated_func, '__wrapped__'):
# Recursion: unwrap more if needed
return cls._unwrap_func(decorated_func.__wrapped__)
else:
# decorated_func isn't actually decorated, no more
# unwrapping to do
return decorated_func
def custom_sort(param):
"""Custom Click(Command|Group).params sorter.
Case insensitive sort with capitals after lowercase. --version at the end since I can't sort --help.
:param click.core.Option param: Parameter to evaluate.
:return: Sort weight.
:rtype: int
"""
option = param.opts[0].lstrip('-')
if param.param_type_name != 'option':
return False,
return True, option == 'version', option.lower(), option.swapcase()
def get_command(self, ctx, command_name):
# if command_name == "BREAK":
# return click.Command("BREAK", help="marker")
options_list = self.get_commands()[command_name]["options"]
key_map = self.commands[command_name]["key_map"]
tasks = self.commands[command_name]["tasks"]
task_vars = self.commands[command_name]["vars"]
default_vars = self.commands[command_name]["default_vars"]
doc = self.commands[command_name]["doc"]
args_that_are_vars = self.commands[command_name]["args_that_are_vars"]
value_vars = self.commands[command_name]["value_vars"]
metadata = self.commands[command_name]["metadata"]
def command_callback(**kwargs):
new_args, final_vars = get_vars_from_cli_input(kwargs, key_map, task_vars, default_vars, args_that_are_vars,
value_vars)
return {"name": command_name, "vars": final_vars, "metadata": metadata}
help = doc.get("help", "n/a")
short_help = doc.get("short_help", help)
epilog = doc.get("epilog", None)
command = click.Command(command_name, params=options_list, help=help, short_help=short_help, epilog=epilog,
callback=command_callback)
return command
def get_command(self, ctx, name):
# if name == BREAK_COMMAND_NAME:
# def break_callback(**kwargs):
# return {"name": BREAK_COMMAND_NAME}
# return click.Command(BREAK_COMMAND_NAME, help="marker to start a new run", callback=break_callback)
if name in self.profile_repo.get_profiles().keys():
return self.profile_repo.get_command(ctx, name)
else:
return None
def generate_cli(spec):
origin_url = None
if isinstance(spec, str):
if spec.startswith('https://') or spec.startswith('http://'):
origin_url = spec
r = requests.get(spec)
r.raise_for_status()
spec = yaml.safe_load(r.text)
else:
with open(spec, 'rb') as fd:
spec = yaml.safe_load(fd.read())
spec = sanitize_spec(spec)
cli = clickclick.AliasedGroup(context_settings=CONTEXT_SETTINGS)
spec = Spec.from_dict(spec, origin_url=origin_url)
for res_name, res in spec.resources.items():
grp = clickclick.AliasedGroup(normalize_command_name(res_name), short_help='Manage {}'.format(res_name))
cli.add_command(grp)
for op_name, op in res.operations.items():
name = get_command_name(op)
cmd = click.Command(name, callback=partial(invoke, op=op), short_help=op.op_spec.get('summary'))
for param_name, param in op.params.items():
if param.required:
arg = click.Argument([param.name])
cmd.params.append(arg)
else:
arg = click.Option(['--' + param.name])
cmd.params.append(arg)
grp.add_command(cmd)
return cli
def command(name=None, cls=None, **kwargs):
available_for = kwargs.pop('available_for', DEFAULT_ACCESS)
if cls is None:
cls = click.Command
def decorator(f):
cmd = click.decorators._make_command(f, name, kwargs, cls)
cmd.__doc__ = f.__doc__
cmd.available_for = available_for
return cmd
return decorator
