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()
python类get_current_context()的实例源码
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)
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
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)
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)
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}
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)
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)
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))
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
def info(*args):
if click.get_current_context().meta['output-logmsg'] != 'silent':
click.secho(*args, err=True, fg='green')
def dmsg(*args):
if click.get_current_context().meta['output-logmsg'] == 'debug':
click.secho(*args, err=True, fg='yellow')
def _fail(message, retcode=1):
click.echo(message, sys.stderr)
click.get_current_context().find_root().exit(retcode)
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)
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
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)
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)
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)
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'
def data(self):
return click.get_current_context().params['config']
