python类get_current_context()的实例源码

options.py 文件源码 项目:globus-cli 作者: globus 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def is_verbose():
    """
    Only safe to call within a click context.
    """
    ctx = click.get_current_context()
    state = ctx.ensure_object(CommandState)
    return state.is_verbose()
pause_info.py 文件源码 项目:globus-cli 作者: globus 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def task_pause_info(task_id):
    """
    Executor for `globus task pause-info`
    """
    client = get_client()
    res = client.task_pause_info(task_id)

    def _custom_text_format(res):
        explicit_pauses = [
            field for field in EXPLICIT_PAUSE_MSG_FIELDS
            # n.b. some keys are absent for completed tasks
            if res.get(field[1])
        ]
        effective_pause_rules = res['pause_rules']

        if not explicit_pauses and not effective_pause_rules:
            safeprint('Task {} is not paused.'.format(task_id))
            click.get_current_context().exit(0)

        if explicit_pauses:
            formatted_print(
                res, fields=explicit_pauses, text_format=FORMAT_TEXT_RECORD,
                text_preamble='This task has been explicitly paused.\n',
                text_epilog='\n' if effective_pause_rules else None)

        if effective_pause_rules:
            formatted_print(
                effective_pause_rules, fields=PAUSE_RULE_DISPLAY_FIELDS,
                text_preamble=(
                    'The following pause rules are effective on this task:\n'))

    formatted_print(res, text_format=_custom_text_format)
helpers.py 文件源码 项目:globus-cli 作者: globus 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def resolve_id_or_name(client, bookmark_id_or_name):
    # leading/trailing whitespace doesn't make sense for UUIDs and the Transfer
    # service outright forbids it for bookmark names, so we can strip it off
    bookmark_id_or_name = bookmark_id_or_name.strip()

    res = None
    try:
        UUID(bookmark_id_or_name)  # raises ValueError if argument not a UUID
    except ValueError:
        pass
    else:
        try:
            res = client.get_bookmark(bookmark_id_or_name.lower())
        except TransferAPIError as exception:
            if exception.code != 'BookmarkNotFound':
                raise

    if not res:  # non-UUID input or UUID not found; fallback to match by name
        try:
            # n.b. case matters to the Transfer service for bookmark names, so
            # two bookmarks can exist whose names vary only by their case
            res = next(bookmark_row for bookmark_row in client.bookmark_list()
                       if bookmark_row['name'] == bookmark_id_or_name)

        except StopIteration:
            safeprint(u'No bookmark found for "{}"'.format(
                      bookmark_id_or_name), write_to_stderr=True)
            click.get_current_context().exit(1)

    return res
shell_completion.py 文件源码 项目:globus-cli 作者: globus 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def do_bash_complete():
    comp_words, quoted = safe_split_line(os.environ['COMP_WORDS'])
    cur_index = int(os.environ['COMP_CWORD'])
    try:
        cur = comp_words[cur_index]
        completed_args = comp_words[1:-1]
    except IndexError:
        cur = None
        completed_args = comp_words[1:]

    choices = get_all_choices(completed_args, cur, quoted)

    safeprint('\t'.join(choices), newline=False)
    click.get_current_context().exit(0)
excepthook.py 文件源码 项目:globus-cli 作者: globus 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def exit_with_mapped_status(http_status):
    """
    Given an HTTP Status, exit with either an error status of 1 or the
    status mapped by what we were given.
    """
    # get the mapping by looking up the state and getting the mapping attr
    mapping = (click.get_current_context().ensure_object(CommandState)
               .http_status_map)

    # if there is a mapped exit code, exit with that. Otherwise, exit 1
    if http_status in mapping:
        sys.exit(mapping[http_status])
    else:
        sys.exit(1)
helper.py 文件源码 项目:twtxt 作者: buckket 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def generate_user_agent():
    try:
        version = pkg_resources.require("twtxt")[0].version
    except pkg_resources.DistributionNotFound:
        version = "unknown"

    conf = click.get_current_context().obj["conf"]
    if conf.disclose_identity and conf.nick and conf.twturl:
        user_agent = "twtxt/{version} (+{url}; @{nick})".format(
            version=version, url=conf.twturl, nick=conf.nick)
    else:
        user_agent = "twtxt/{version}".format(version=version)

    return {"User-Agent": user_agent}
mentions.py 文件源码 项目:twtxt 作者: buckket 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def get_source_by_url(url):
    conf = click.get_current_context().obj["conf"]
    if url == conf.twturl:
        return conf.source
    return next((source for source in conf.following if url == source.url), None)
mentions.py 文件源码 项目:twtxt 作者: buckket 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def get_source_by_name(nick):
    nick = nick.lower()
    conf = click.get_current_context().obj["conf"]
    if nick == conf.nick and conf.twturl:
        return conf.source
    return next((source for source in conf.following if nick == source.nick), None)
cli.py 文件源码 项目:histonets-cv 作者: sul-cidr 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def main(rst=None):
    """Histonets computer vision application for image processing"""
    ctx = click.get_current_context()
    if rst:
        click.echo()
        comamnds_text = 'Commands'
        options_text = 'Options:'
        main_help, _ = main.get_help(ctx).split(comamnds_text, 1)
        click.echo(main_help)
        click.echo(comamnds_text)
        click.echo('-' * len(comamnds_text))
        click.echo()
        for command_name, command in sorted(main.commands.items()):
            click.echo(command_name)
            click.echo('~' * len(command_name))
            click.echo(command.get_usage(ctx)
                              .replace('histonets ', "histonets {} ".format(
                                       command_name))
                              .replace('\b\n', ''))
            click.echo()
            click.echo(command.help.replace('\b\n', ''))
            command_help = command.get_help(ctx)
            _, command_options_help = command_help.split(options_text, 1)
            command_options, _ = command_options_help.rsplit('--help', 1)
            click.echo()
            click.echo(options_text)
            click.echo(command_options)
            click.echo()
    elif ctx.invoked_subcommand is None:
        click.echo(main.get_help(ctx))
cli.py 文件源码 项目:histonets-cv 作者: sul-cidr 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def pipeline(image, actions):
    """Allow chaining a series of actions to be applied to IMAGE.
    Output will depend on the last action applied.

    \b
    - ACTIONS is a JSON list of dictionaries containing each an 'action' key
      specifying the action to apply, a 'arguments' key which is a
      list of arguments, and a 'options' key with a dictionary to set the
      options for the corresponding action.

      Example::

        histonets pipeline '[{"action": "contrast", "options": {"value": 50}}]'
    """
    output = image.image
    for action in actions:
        ctx = click.get_current_context()
        arguments = [output] + action.get('arguments', [])
        command = main.get_command(ctx, action['action'])
        if command is None:
            raise click.BadParameter(
                "Action '{}' not found".format(action['action']))
        action_options = action.get('options', {})
        options = {param.name: action_options.get(param.name, param.default)
                   for param in command.params[:-2]}
        options['output'] = RAW
        try:
            output = command.callback(*arguments, **options)
        except TypeError as e:
            raise click.BadParameter(e)
    return output
cli.py 文件源码 项目:ndeftool 作者: nfcpy 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
def info(*args):
    if click.get_current_context().meta['output-logmsg'] != 'silent':
        click.secho(*args, err=True, fg='green')
cli.py 文件源码 项目:ndeftool 作者: nfcpy 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
def dmsg(*args):
    if click.get_current_context().meta['output-logmsg'] == 'debug':
        click.secho(*args, err=True, fg='yellow')
cli.py 文件源码 项目:databrewer 作者: rmax 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _fail(message, retcode=1):
    click.echo(message, sys.stderr)
    click.get_current_context().find_root().exit(retcode)
cli.py 文件源码 项目:databrewer 作者: rmax 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def _echo(output, min_lines=10):
    ctx = click.get_current_context()
    if ctx.obj.get('use_pager') and output.count('\n') > min_lines:
        _func = click.echo_via_pager
    else:
        _func = click.echo
    _func(output, sys.stdout)
__main__.py 文件源码 项目:kozinaki 作者: compunova 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
def cmd_callback(self, name):
        ctx = click.get_current_context()
        for node_name in name:
            node = node_manager.node_get(node_name=node_name)
            if node:
                response = node.command(cmd=ctx.command.name)
                for r in response:
                    print(r)
            else:
                click.secho('Node "{}" not found'.format(node_name), fg='red')


# CREATE PROVIDER NODES
__main__.py 文件源码 项目:kozinaki 作者: compunova 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def create_node_callback(self, name, compute_driver, **kwargs):
        with click_spinner.spinner():
            ctx = click.get_current_context()
            node_manager.node_create(node_name=name, node_type=ctx.command.name,
                                     compute_driver=compute_driver, **kwargs)
query.py 文件源码 项目:anchore 作者: anchore 项目源码 文件源码 阅读 32 收藏 0 点赞 0 评论 0
def init_nav_contexts():
    try:
        # use the obj from the current click context. This is a bit hacky, but works as long as this method is
        # invoked in an execution context of click
        anchore_config = click.get_current_context().obj
        nav = navigator.Navigator(anchore_config=anchore_config, imagelist=imagelist, allimages=contexts['anchore_allimages'])
        return nav
    except Exception as err:
        anchore_print_err("explore operation failed")
        success = False
        ecode = 1

    if not success:
        contexts['anchore_allimages'].clear()
        sys.exit(ecode)
audit.py 文件源码 项目:anchore 作者: anchore 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
def init_nav_contexts():
    try:
        # use the obj from the current click context. This is a bit hacky, but works as long as this method is
        # invoked in an execution context of click
        anchore_config = click.get_current_context().obj
        nav = navigator.Navigator(anchore_config=anchore_config, imagelist=imagelist, allimages=contexts['anchore_allimages'])
        return nav
    except Exception as err:
        anchore_print_err("explore operation failed")
        success = False
        ecode = 1

    if not success:
        contexts['anchore_allimages'].clear()
        sys.exit(ecode)
settings.py 文件源码 项目:Sentry 作者: NetEaseGame 项目源码 文件源码 阅读 25 收藏 0 点赞 0 评论 0
def get_sentry_conf():
    """
    Fetch the SENTRY_CONF value, either from the click context
    if available, or SENTRY_CONF environment variable.
    """
    try:
        ctx = click.get_current_context()
        return ctx.obj['config']
    except (RuntimeError, KeyError):
        try:
            return os.environ['SENTRY_CONF']
        except KeyError:
            return '~/.sentry'
config.py 文件源码 项目:logstapo 作者: ThiefMaster 项目源码 文件源码 阅读 28 收藏 0 点赞 0 评论 0
def data(self):
        return click.get_current_context().params['config']


问题


面经


文章

微信
公众号

扫码关注公众号