Python click 模块，Command() 实例源码

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

def main():
    # type: () -> click.Command
    return cli(obj={})

def _format_description(ctx):
    """Format the description for a given `click.Command`.

    We parse this as reStructuredText, allowing users to embed rich
    information in their help messages if they so choose.
    """
    if not ctx.command.help:
        return

    for line in statemachine.string2lines(
            ctx.command.help, tab_width=4, convert_whitespace=True):
        yield line
    yield ''

def _format_usage(ctx):
    """Format the usage for a `click.Command`."""
    yield '.. code-block:: shell'
    yield ''
    for line in _get_usage(ctx).splitlines():
        yield _indent(line)
    yield ''

def _format_options(ctx):
    """Format all `click.Option` for a `click.Command`."""
    # the hidden attribute is part of click 7.x only hence use of getattr
    params = [x for x in ctx.command.params if isinstance(x, click.Option)
              and not getattr(x, 'hidden', False)]

    for param in params:
        for line in _format_option(param):
            yield line
        yield ''

def _format_arguments(ctx):
    """Format all `click.Argument` for a `click.Command`."""
    params = [x for x in ctx.command.params if isinstance(x, click.Argument)]

    for param in params:
        for line in _format_argument(param):
            yield line
        yield ''

def _format_envvars(ctx):
    """Format all envvars for a `click.Command`."""
    params = [x for x in ctx.command.params if getattr(x, 'envvar')]

    for param in params:
        for line in _format_envvar(param):
            yield line
        yield ''

def __init__(self, *args, **kwargs):
        cs = dict(help_option_names=['-h', '--help'])
        super(Command, self).__init__(context_settings=cs,
                                      *args, **kwargs)


# Automatically load commands from the cli folder

def cli(main, conf_dir=None, commands_dir=None):
    """Convenience function for initialising a Command CLI

    For parameter definitions see :class:`.Command`
    """
    return Command(main, conf_dir=conf_dir, commands_dir=commands_dir)()

def _get_commands_from_prefix_methods(self, method_prefix):
        for method in self._get_prefix_methods(method_prefix):
            result = method()
            if result is not None and isinstance(result, click.Command):
                yield result

def simulation_commands():
    """Make commands simulations imported from python files in current working 
    directory."""
    simulations = import_simulations(dir_path='.')
    for simulation_name, simulation_cls in simulations.items():
        # New command
        command = click.Command(
            simulation_name, callback=lambda: simulation_cls().run())

        # Add the command into run group
        run.add_command(command)

        # Add options for setting up the simulation
        for name, trait in class_own_traits(simulation_cls):
            command.params.append(trait_to_option(name, trait))

def create_cmd(self, cmd_action):

        option_name = click.Argument(['name'], nargs=-1, required=True)

        cmd = click.Command(
            name=cmd_action,
            params=[option_name],
            help="{} all node's services".format(cmd_action.capitalize()),
            callback=self.cmd_callback
        )

        return cmd

def main(verbose):
    """Command-line interface to manage cloud provider nodes for the Kozinaki driver.
    A node is a collection of OpenStack service instances responsible for interaction with the cloud provider's API.
    There could be multiple nodes created for the same cloud provider with different parameters.
    """
    pass

def get_command(self, ctx, cmd_name):
        """This function checks whether ``cmd_name`` can be matched to one
        of the registered commands.

        Parameters
        ----------
        cmd_name : str
            Input string from the command line.

        Returns
        -------
        matched_command : click.Command
            Returns matched command

        """
        matched_command = click.Group.get_command(self, ctx, cmd_name)
        if matched_command is not None:
            return matched_command
        matches = [x for x in self.list_commands(ctx)
                   if x.startswith(cmd_name)]
        if not matches:
            return None
        elif len(matches) == 1:
            matched_command = click.Group.get_command(self, ctx, matches[0])
            return matched_command
        ctx.fail('Too many matches: %s' % ', '.join(sorted(matches)))

def __init__(self, command, description, parameters, function, plugin):
        self.command = command
        self.description = description
        self.parameters = parameters
        self.plugin = plugin
        self.function = function
        self.click_command = click.Command(command, callback=function, help=description, params=parameters)

def command(name=None, cls=None, **attrs):
    """
    Wrapper for click Commands, to replace the click.Command docstring with the
    docstring of the wrapped method (i.e. the methods defined below). This is
    done to support the autodoc in Sphinx, and the correct display of
    docstrings
    """
    if cls is None:
        cls = Command

    def decorator(f):
        r = _make_command(f, name, attrs, cls)
        r.__doc__ = f.__doc__
        return r
    return decorator

def command(name=None, cls=None, **attrs):
    """
    Wrapper for click Commands, to replace the click.Command docstring with the
    docstring of the wrapped method (i.e. the methods defined below). This is
    done to support the autodoc in Sphinx, and the correct display of docstrings
    """
    if cls is None:
        cls = Command
    def decorator(f):
        r = _make_command(f, name, attrs, cls)
        r.__doc__ = f.__doc__
        return r
    return decorator

def _generate_command_reply(cmd):
    """Recursively generate completion reply for this command and subcommands.

    Parameters
    ----------
    cmd : click.Command
        Command to generate completion replies for (including its subcommands).

    """
    import textwrap
    import click

    ctx = None

    options = ['--help']
    for param in cmd.params:
        if isinstance(param, click.Option):
            options.extend(param.opts)
            options.extend(param.secondary_opts)

    subcmd_names = []
    if isinstance(cmd, click.MultiCommand):
        subcmd_names.extend(cmd.list_commands(ctx))

    subcmd_cases = []
    for subcmd_name in subcmd_names:
        subcmd_reply = _generate_command_reply(
            cmd.get_command(ctx, subcmd_name))
        subcmd_reply = textwrap.indent(subcmd_reply, '  ')

        case = SUBCOMMAND_CASE_TEMPLATE.format(
            subcmd_name=subcmd_name, subcmd_reply=subcmd_reply)
        subcmd_cases.append(case)

    subcmd_cases = textwrap.indent('\n'.join(subcmd_cases), ' ' * 6)

    cmd_reply = COMMAND_REPLY_TEMPLATE.format(
        options=' '.join(options), subcmd_names=' '.join(subcmd_names),
        subcmd_cases=subcmd_cases)

    return cmd_reply


# NOTE: using double braces to avoid `str.format` interpolation when bash needs
# curly braces in the generated code.
#
# NOTE: the handling of a negative COMP_CWORD is necessary in certain versions
# of bash (e.g. at least the bash shipped with OS X 10.9.5). When adding
# whitespace to the end of a command, and then moving the cursor backwards in
# the command and hitting <tab>, COMP_CWORD can be negative (I've only seen -2
# as its value). This is a bash bug and is not documented behavior. Other CLIs
# with tab completion suffer from the same issue, and each one deals with this
# bug differently (some not at all, e.g. `git`). The workaround used below
# seems to provide the least destructive completion behavior for our CLI.
#
# Bug report reference:
#   https://lists.gnu.org/archive/html/bug-bash/2009-07/msg00108.html

def get_completion_context(args):
    """
    Walk the tree of commands to a terminal command or multicommand, using the
    Click Context system.
    Effectively, we'll be using the resilient_parsing mode of commands to stop
    evaluation, then having them capture their options and arguments, passing
    us on to the next subcommand. If we walk "off the tree" with a command that
    we don't recognize, we have a hardstop condition, but otherwise, we walk as
    far as we can go and that's the location from which we should do our
    completion work.
    """
    # get the "globus" command as a click.Command
    root_command = click.get_current_context().find_root().command
    # build a new context object off of it, with resilient_parsing set so that
    # no callbacks are invoked
    ctx = root_command.make_context('globus', list(args),
                                    resilient_parsing=True)

    # walk down multicommands until we've matched on everything and are at a
    # terminal context that holds all of our completed args
    while isinstance(ctx.command, click.MultiCommand) and args:
        # trim out any params that are capturable at this level of the command
        # tree by resetting the argument list
        args = ctx.protected_args + ctx.args

        # if there were no remaining args, stop walking the tree
        if not args:
            break

        # check for a matching command, and if one isn't found stop the
        # traversal and abort the whole process -- this would mean that a
        # completed command was entered which doesn't match a known command
        # there's nothing completion can do in this case unless it implements
        # sophisticated fuzzy matching
        command = ctx.command.get_command(ctx, args[0])
        if not command:
            return None

        # otherwise, grab that command, and build a subcontext to continue the
        # tree walk
        else:
            ctx = command.make_context(args[0], args[1:], parent=ctx,
                                       resilient_parsing=True)

    # return the context we found
    return ctx

def _format_command(ctx, show_nested):
    """Format the output of `click.Command`."""

    # description

    for line in _format_description(ctx):
        yield line

    yield '.. program:: {}'.format(ctx.command_path)

    # usage

    for line in _format_usage(ctx):
        yield line

    # options

    lines = list(_format_options(ctx))
    if lines:
        # we use rubric to provide some separation without exploding the table
        # of contents
        yield '.. rubric:: Options'
        yield ''

    for line in lines:
        yield line

    # arguments

    lines = list(_format_arguments(ctx))
    if lines:
        yield '.. rubric:: Arguments'
        yield ''

    for line in lines:
        yield line

    # environment variables

    lines = list(_format_envvars(ctx))
    if lines:
        yield '.. rubric:: Environment variables'
        yield ''

    for line in lines:
        yield line

    # if we're nesting commands, we need to do this slightly differently
    if show_nested:
        return

    commands = sorted(getattr(ctx.command, 'commands', {}).items())

    if commands:
        yield '.. rubric:: Commands'
        yield ''

    for command in commands:
        for line in _format_subcommand(command):
            yield line
        yield ''

def _generate_nodes(self, name, command, parent=None, show_nested=False):
        """Generate the relevant Sphinx nodes.

        Format a `click.Group` or `click.Command`.

        :param name: Name of command, as used on the command line
        :param command: Instance of `click.Group` or `click.Command`
        :param parent: Instance of `click.Context`, or None
        :param show_nested: Whether subcommands should be included in output
        :returns: A list of nested docutil nodes
        """
        ctx = click.Context(command, info_name=name, parent=parent)

        # Title

        section = nodes.section(
            '',
            nodes.title(text=name),
            ids=[nodes.make_id(ctx.command_path)],
            names=[nodes.fully_normalize_name(ctx.command_path)])

        # Summary

        source_name = ctx.command_path
        result = statemachine.ViewList()

        lines = _format_command(ctx, show_nested)
        for line in lines:
            result.append(line, source_name)

        self.state.nested_parse(result, 0, section)

        # Subcommands

        if show_nested:
            commands = getattr(ctx.command, 'commands', {})
            for command_name, command_obj in sorted(commands.items()):
                section.extend(self._generate_nodes(
                    command_name,
                    command_obj,
                    ctx,
                    show_nested))

        return [section]

def create_command(self, provider_name):
        config = node_manager.valid_node_types['providers'].get(provider_name)

        if not config:
            return

        provider_options = [
            click.Option(param_decls=['--name'], help='Compute node name', required=True)
        ]

        for param, param_data in node_manager.get_node_params(provider_name).items():
            argument_params = dict()
            argument_params['help'] = ''

            description = param_data.get('description', {}).get(DEFAULT_LANG)
            default = param_data.get('default')
            arg_type = param_data.get('type')

            if description:
                argument_params['help'] += '{} '.format(description)
            if arg_type:
                argument_params['help'] += '(type: {}) '.format(arg_type)
            if default:
                argument_params.update({
                    'default': default,
                    'required': False
                })
                argument_params['help'] += '(default: {}) '.format(default)
            else:
                argument_params['required'] = True

            provider_options.append(click.Option(
                param_decls=['--{}'.format(param)],
                help=argument_params['help'],
                default=default,
                required=False if default else True
            ))

        cmd = click.Command(
            name=provider_name,
            params=provider_options,
            help=click.style(config['description'], fg='cyan'),
            short_help=click.style(config['description'], fg='cyan'),
            callback=self.create_node_callback
        )

        return cmd