python类argument()的实例源码

def mkdir(directory, exists_okay):
    """
    Create a directory on the board.

    Mkdir will create the specified directory on the board.  One argument is
    required, the full path of the directory to create.

    Note that you cannot recursively create a hierarchy of directories with one
    mkdir command, instead you must create each parent directory with separate
    mkdir command calls.

    For example to make a directory under the root called 'code':

      ampy --port /board/serial/port mkdir /code
    """
    # Run the mkdir command.
    board_files = files.Files(_board)
    board_files.mkdir(directory,
                      exists_okay=exists_okay)

def ls(directory):
    """List contents of a directory on the board.

    Can pass an optional argument which is the path to the directory.  The
    default is to list the contents of the root, /, path.

    For example to list the contents of the root run:

      ampy --port /board/serial/port ls

    Or to list the contents of the /foo/bar directory on the board run:

      ampy --port /board/serial/port ls /foo/bar
    """
    # List each file/directory on a separate line.
    board_files = files.Files(_board)
    for f in board_files.ls(directory):
        print(f)

def execute_if_needed(argument, spec, use_cache):
    ps = status.get(spec.pipeline_id)
    if (match_pipeline_id(argument, spec.pipeline_id) or
            (argument == 'all') or
            (argument == 'dirty' and ps.dirty())):
        if len(spec.validation_errors) != 0:
            return (spec.pipeline_id, False, {}, ['init'] + list(map(str, spec.validation_errors))), \
                   spec.pipeline_id == argument
        eid = gen_execution_id()
        if ps.queue_execution(eid, 'manual'):
            success, stats, errors = \
                execute_pipeline(spec, eid,
                                 use_cache=use_cache)
            return (spec.pipeline_id, success, stats, errors), spec.pipeline_id == argument
        else:
            return (spec.pipeline_id, False, None, ['Already Running']), spec.pipeline_id == argument
    return None, False

def runserver(**config):
    """
    Run a development server for an aiohttp apps.

    Takes one argument "app-path" which should be a path to either a directory containing a recognized default file
    ("app.py" or "main.py") or to a specific file. Defaults to the environment variable "AIO_APP_PATH" or ".".

    The app path is run directly, see the "--app-factory" option for details on how an app is loaded from a python
    module.
    """
    active_config = {k: v for k, v in config.items() if v is not None}
    setup_logging(config['verbose'])
    try:
        run_app(*_runserver(**active_config))
    except AiohttpDevException as e:
        if config['verbose']:
            tb = click.style(traceback.format_exc().strip('\n'), fg='white', dim=True)
            main_logger.warning('AiohttpDevException traceback:\n%s', tb)
        main_logger.error('Error: %s', e)
        sys.exit(2)

def handle_jinja_errors(func):
    '''
    Print nice error messages on jinja exceptions.

    Expect a *base_dir* argument, so normally used after ``@repos_command``.
    '''

    @functools.wraps(func)
    def wrapper(base_dir, **kwargs):
        try:
            return func(base_dir=base_dir, **kwargs)
        except jinja2.TemplateNotFound as exc:
            click.secho('Template not found: %s' % exc.name, fg='red',
                        bold=True)
        except jinja2.TemplateError:
            exc_info = sys.exc_info()
            stack, error = templates.extract_jinja_error(exc_info, base_dir)
            templates.print_jinja_error(stack, error)
        sys.exit(1)

    return wrapper

def package_args(versioned=True):
    if versioned:
        def wrapper(f):
            return all_opt(sort_opt(pre_opt(click.argument(
                'packages',
                nargs=-1,
                callback=lambda ctx, param, value:
                            ctx.obj.lookup_package_version(value),
            )(f))))
        return wrapper
    else:
        return click.argument(
            'packages',
            nargs=-1,
            callback=lambda ctx, param, value: ctx.obj.lookup_package(value),
        )

def make_django_command(name, django_command=None, help=None):
    "A wrapper to convert a Django subcommand a Click command"
    if django_command is None:
        django_command = name

    @click.command(
        name=name,
        help=help,
        add_help_option=False,
        context_settings=dict(
            ignore_unknown_options=True,
        ))
    @click.argument('management_args', nargs=-1, type=click.UNPROCESSED)
    @click.pass_context
    def inner(ctx, management_args):
        from sentry.runner.commands.django import django
        ctx.params['management_args'] = (django_command,) + management_args
        ctx.forward(django)

    return inner

def set_help_from_yaml(cls, ctx, param, value):
        """
        When attaching this function as `callback` argument to an Option (`click.Option`),
        it will set
        an automatic help for all Options of the same command, which do not have an `help`
        specified and are found in the default config file for downloading
        (currently `download.yaml`).
        The Option having as callback this function must also have `is_eager=True`.
        Example:
        Assuming opt1, opt2, opt3 are variables of the config yaml file, and opt4 not, this
        sets the default help for opt1 and opt2:

def download(configfile, dburl, eventws, start, end, dataws, min_sample_rate, traveltimes_model,
             timespan, update_metadata, retry_url_err, retry_mseed_err, retry_seg_not_found,
             retry_client_err, retry_server_err, retry_timespan_err, inventory, eventws_query_args):
    """Download waveform data segments with quality metadata and relative events, stations and
    channels metadata into a specified database.
    The -c option (required) sets the defaults for all other options below, which are optional.
    The argument 'eventws_query_args' is an optional list of space separated key and values to be
    passed to the event web service query (example: minmag 5.5 minlon 34.5). All FDSN query
    arguments are valid except 'start', 'end' (set them via -t0 and -t1) and 'format'
    """
    try:
        cfg_dict = yaml_load(configfile, **{k: v for k, v in locals().items()
                                            if v not in ((), {}, None, configfile)})
        # start and end might be integers. If we attach the conversion function
        # `clickutils.valid_date` to the relative clikc Option 'type' argument, the
        # function does not affect integer values in the config. Thus we need to set it here:
        cfg_dict['start'] = clickutils.valid_date(cfg_dict['start'])
        cfg_dict['end'] = clickutils.valid_date(cfg_dict['end'])
        ret = main.download(isterminal=True, **cfg_dict)
        sys.exit(ret)
    except KeyboardInterrupt:  # this except avoids printing traceback
        sys.exit(1)  # exit with 1 as normal python exceptions

def argument_list(cls, name):
        """
        A `click.argument` decorator that accepts zero or more command-line
        arguments and converts them all to instances of the class.  If
        `default_params` is not `None`, an empty argument list will be
        defaulted to the resource type's default value.
        """
        if cls.default_params is not None:
            def callback(ctx, param, value):
                return value or [cls.default(ctx.obj)]
        else:
            callback = None
        return click.argument(
            name,
            type=ResourceParamType(cls),
            nargs=-1,
            callback=callback,
        )

def construct_action(self, instance, spec, loc, context, action):
        """
        Construct a command based on a specific actions, e.g. list,
        create, etc.
        """
        assert len(loc) == 4
        self.init_adapter_conf(instance)
        collection = loc[0] + '/' + loc[-3]
        command = self.COMMANDS[action](self.clients.get(collection))
        if action in self.RESOURCE_ACTIONS:
            command = click.argument('resource_id')(command)
        if action in self.CRITICAL_ACTIONS:
            option = click.option(
                '--yes', is_flag=True, callback=abort_if_false,
                expose_value=False,
                prompt='Are you sure you want to perform this action?')
            command = option(command)
        self._add_format_option(command, action)
        instance[self.ADAPTER_CONF][action] = command
        return instance

def main(number=None):
    """Console script for rust_pypi_example
       The console script takes a singe argument, "NUMBER",
       which must be an integer greater than 2. The script calls
       out to a library written in Rust to compute whether the
       intger is a prime number.
       Example:
           rust_pypi_example 13
    """
    if number and number > 2:
        click.echo(True if rust_lib.is_prime(number) else False)
    else:
        click.echo("Please supply an integer argument greater than 2. The "
                   "console script will tell you if it is a prime number")

def update(project, name, description):
    """Update project.

    Example:

def update(group, project, description):
    """Update experiement group.

    Example:

def cli(username, repository, path, password, deploy, env):
    """Encrypt passwords and environment variables for use with Travis CI.

    Travis Encrypt requires as arguments the user's GitHub username and repository name.
    Once the arguments are passed, a password prompt will ask for the password that needs
    to be encrypted. The given password will then be encrypted via the PKCS1v15 padding
    scheme and printed to standard output. If the path to a .travis.yml file
    is given as an argument, the encrypted password is added to the .travis.yml file.
    """
    key = retrieve_public_key('{}/{}' .format(username, repository))
    encrypted_password = encrypt_key(key, password.encode())

    if path:
        config = load_travis_configuration(path)

        if deploy:
            config.setdefault('deploy', {}).setdefault('password', {})['secure'] = encrypted_password

        elif env:
            config.setdefault('env', {}).setdefault('global', {})['secure'] = encrypted_password

        else:
            config.setdefault('password', {})['secure'] = encrypted_password

        dump_travis_configuration(config, path)

        print('Encrypted password added to {}' .format(path))

    else:
        print('\nPlease add the following to your .travis.yml:\nsecure: {}' .format(encrypted_password))

def updateteam(command, teamid, userkey):
    """Update a team by adding or deleting a user."""
    if command == 'add_user':
        api.add_user_to_team(teamid, userkey)
    elif command == 'delete_user':
        api.delete_user_from_team(teamid, userkey)
    else:
        click.echo('Missing argument "command".')

def _parse_args_and_kwargs(passed_args):

    args = []
    kwargs = {}

    has_kwargs = False

    while len(passed_args) > 0:

        arg = passed_args.pop(0)

        if arg[:2] == '--':
            has_kwargs = True

            if not len(passed_args) > 0:
                raise ValueError('Argument "{}" not recognized'.format(arg))

            kwargs[arg[2:]] = passed_args.pop(0)

        else:
            if has_kwargs:
                raise ValueError(
                    'Positional argument "{}" after keyword arguments'.format(
                        arg))

            args.append(arg)

    return args, kwargs

def convert(self, value, param, ctx):
        try:
            if isinstance(value, (int,)):
                return value
            else:
                return int(value, 0)
        except:
            self.fail('%s is not a valid value' % value, param, ctx)


# Create instances of custom argument types

def convert(self, value, param, ctx):
        try:
            if isinstance(value, (int,)):
                return value
            else:
                return int(value, 0)
        except:
            self.fail('%s is not a valid value' % value, param, ctx)


# Create instances of custom argument types

def list():
    for name, config in get_config()['tunnels'].items():
        print name
        # print {'name':name,
        #     'proxy': config['proxy']['host'] if 'proxy' in config and 'host' in config['proxy'] else '',
        #     'check': config['check'] if 'check' in config else '',
        #     'depends':config['depends'] if 'depends' in config else ''}


# @cli.command()
# @click.argument('tunnel')
# def up(tunnel):
#     try:
#         if not supervisor_is_running():
#             start()
#
#         get_supervisor().startProcess(tunnel)
#     except:
#         traceback.print_exc()


# @cli.command('down')
# @click.argument('tunnel')
# def down(tunnel):
#     if not supervisor_is_running():
#         print('supervisor is not running')
#
#     get_config(None)
#     write_supervisor_conf()
#
#     get_supervisor().stopProcess(tunnel)
#     # continue stopping recursively
#     stop_dependent_tunnels(tunnel)
#
#     # get_supervisor().stopProcess(tunnel)

def get_choices(cli, prog_name, args, incomplete):
    ctx = resolve_ctx(cli, prog_name, args)
    if ctx is None:
        return

    optctx = None
    if args:
        for param in ctx.command.get_params(ctx):
            if isinstance(param, Option) and not param.is_flag and args[-1] in param.opts + param.secondary_opts:
                optctx = param

    choices = []
    if optctx:
        choices += [c if isinstance(c, tuple) else (c, None) for c in optctx.type.complete(ctx, incomplete)]
    elif incomplete and not incomplete[:1].isalnum():
        for param in ctx.command.get_params(ctx):
            if not isinstance(param, Option):
                continue
            for opt in param.opts:
                if startswith(opt, incomplete):
                    choices.append((opt, param.help))
            for opt in param.secondary_opts:
                if startswith(opt, incomplete):
                    # don't put the doc so fish won't group the primary and
                    # and secondary options
                    choices.append((opt, None))
    elif isinstance(ctx.command, MultiCommand):
        for name in ctx.command.list_commands(ctx):
            if startswith(name, incomplete):
                choices.append((name, ctx.command.get_command_short_help(ctx, name)))
    else:
        for param in ctx.command.get_params(ctx):
            if isinstance(param, Argument):
                choices += [c if isinstance(c, tuple) else (c, None) for c in param.type.complete(ctx, incomplete)]

    for item, help in choices:
        yield (item, help)

def activate(self):
        self.commands.register("recipe_list", "Lists all recipes", self._recipe_list)
        self.commands.register("recipe_build", "Builds a given recipe", self._recipe_build,
                               params=[Argument(("recipe",), required=True)])

        self.recipes.register("gw_package",
                              os.path.abspath(os.path.join(os.path.dirname(__file__), "../recipes/gw_package")),
                              description="Groundwork basic package. Includes places for "
                                          "apps, plugins, patterns and recipes.",
                              final_words="Recipe Installation is done.\n\n"
                                          "During development use buildout:\n"
                                          "Run: python bootstrap.py\n"
                                          "Then: bin/buildout\n"
                                          "Start the app: bin/app\n\n"
                                          "For installation run: 'python setup.py install' \n"
                                          "For documentation run: 'make html' inside doc folder "
                                          "(after installation!)\n\n"
                                          "For more information, please take a look into the README file "
                                          "to know how to go on.\n"
                                          "For help visit: https://groundwork.readthedocs.io\n\n"
                                          "Have fun with your groundwork package.")

def get_choices(cli, prog_name, args, incomplete):
    ctx = resolve_ctx(cli, prog_name, args)
    if ctx is None:
        return

    optctx = None
    if args:
        for param in ctx.command.get_params(ctx):
            if isinstance(param, Option) and not param.is_flag and args[-1] in param.opts + param.secondary_opts:
                optctx = param

    choices = []
    if optctx:
        choices += [c if isinstance(c, tuple) else (c, None) for c in optctx.type.complete(ctx, incomplete)]
    elif incomplete and not incomplete[:1].isalnum():
        for param in ctx.command.get_params(ctx):
            if not isinstance(param, Option):
                continue
            for opt in param.opts:
                if startswith(opt, incomplete):
                    choices.append((opt, param.help))
            for opt in param.secondary_opts:
                if startswith(opt, incomplete):
                    # don't put the doc so fish won't group the primary and
                    # and secondary options
                    choices.append((opt, None))
    elif isinstance(ctx.command, MultiCommand):
        for name in ctx.command.list_commands(ctx):
            if startswith(name, incomplete):
                choices.append((name, ctx.command.get_command_short_help(ctx, name)))
    else:
        for param in ctx.command.get_params(ctx):
            if isinstance(param, Argument):
                choices += [c if isinstance(c, tuple) else (c, None) for c in param.type.complete(ctx, incomplete)]

    for item, help in choices:
        yield (item, help)

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 metavar(self):
        """
        Metavar as a property, to satisfy the slight brokenness of
        click.Argument
        Tracked against Click as an issue:
        https://github.com/pallets/click/issues/674
        """
        if self.path_required:
            return "ENDPOINT_ID:PATH"
        else:
            return "ENDPOINT_ID[:PATH]"

def _format_argument(arg):
    """Format the output of a `click.Argument`."""
    yield '.. option:: {}'.format(arg.human_readable_name)
    yield ''
    yield _indent('{} argument{}'.format(
        'Required' if arg.required else 'Optional',
        '(s)' if arg.nargs != 1 else ''))

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_envvar(param):
    """Format the envvars of a `click.Option` or `click.Argument`."""
    yield '.. envvar:: {}'.format(param.envvar)
    yield ''
    if isinstance(param, click.Argument):
        param_ref = param.human_readable_name
    else:
        # if a user has defined an opt with multiple "aliases", always use the
        # first. For example, if '--foo' or '-f' are possible, use '--foo'.
        param_ref = param.opts[0]

    yield _indent('Provide a default for :option:`{}`'.format(param_ref))

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 activate(self):
        """
        Activates GwDocumentsInfo by registering:

        * 2 commands (doc, doc_list)
        * 1 document (documents_overview)
        """
        self.commands.register("doc_list", "List all documents", self._list_documents)
        self.commands.register("doc", "Shows the documentation", self._show_documentation)
        self.commands.register("doc_write", "Stores documents as files ", self._store_documentation,
                               params=[Argument(("path",),
                                                required=True),
                                       Option(("--html", "-h"),
                                              required=False,
                                              help="Will output html instead of rst",
                                              default=False,
                                              is_flag=True),
                                       Option(("--overwrite", "-o"),
                                              required=False,
                                              help="Will overwrite existing files",
                                              default=False,
                                              is_flag=True),
                                       Option(("--quiet", "-q"),
                                              required=False,
                                              help="Will suppress any user interaction",
                                              default=False,
                                              is_flag=True)
                                       ])

        self.documents.register(name="documents_overview",
                                content=documents_content,
                                description="Gives an overview about all registered documents")