Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def custom_result_renderer(res, **kwargs): # pragma: no cover
from datalad.ui import ui
if res.get('action', None) == 'create' and \
res.get('status', None) == 'ok' and \
res.get('type', None) == 'dataset':
ui.message("Created dataset at {}.".format(res['path']))
else:
ui.message("Nothing was created")
def custom_result_renderer(res, **kwargs): # pragma: more cover
from datalad.ui import ui
if res.get('action', None) == 'create' and \
res.get('status', None) == 'ok' and \
res.get('type', None) == 'dataset':
ui.message("Created dataset at {}.".format(res['path']))
else:
ui.message("Nothing was created")
def custom_result_renderer(res, **kwargs):
if res['status'] != 'ok' or not res.get('action', None) == 'metadata':
# logging complained about this already
return
# list the path, available metadata keys, and tags
path = op.relpath(res['path'],
res['refds']) if res.get('refds', None) else res['path']
meta = res.get('metadata', {})
ui.message('{path}{type}:{spacer}{meta}{tags}'.format(
path=ac.color_word(path, ac.BOLD),
type=' ({})'.format(
ac.color_word(res['type'], ac.MAGENTA)) if 'type' in res else '',
spacer=' ' if len([m for m in meta if m != 'tag']) else '',
meta=','.join(k for k in sorted(meta.keys())
if k not in ('tag', '@context', '@id'))
if meta else ' -' if 'metadata' in res else ' aggregated',
tags='' if 'tag' not in meta else ' [{}]'.format(
','.join(assure_list(meta['tag'])))))
# so we knew it needs authentication
if used_old_session:
# Let's try with fresh ones
allow_old_session = False
continue
else:
# we did use new cookies, we knew that authentication is needed
# but still failed. So possible cases:
# 1. authentication credentials changed/were revoked
# - allow user to re-enter credentials
# 2. authentication mechanisms changed
# - we can't do anything here about that
# 3. bug in out code which would render authentication/cookie handling
# ineffective
# - not sure what to do about it
if not ui.is_interactive:
lgr.error(
"Interface is non interactive, so we are "
"reraising: %s" % exc_str(e))
reraise(*exc_info)
self._enter_credentials(
url,
denied_msg=access_denied,
auth_types=supported_auth_types,
new_provider=False)
allow_old_session = False
continue
else: # None or False
if needs_authentication is False:
# those urls must or should NOT require authentication
# but we got denied
raise DownloadError(
def color_word(s, color):
return "%s%s%s" % (COLOR_SEQ % color, s, RESET_SEQ) \
if ui.is_interactive \
else s
def enter_new(self, url=None, auth_types=[]):
from datalad.ui import ui
name = None
if url:
ri = RI(url)
for f in ('hostname', 'name'):
try:
# might need sanitarization
name = str(getattr(ri, f))
except AttributeError:
pass
known_providers_by_name = {p.name: p for p in self._providers}
providers_user_dir = self._get_providers_dirs()['user']
while True:
name = ui.question(
title="New provider name",
text="Unique name to identify 'provider' for %s" % url,
default=name
)
filename = pathjoin(providers_user_dir, '%s.cfg' % name)
if name in known_providers_by_name:
if ui.yesno(
title="Known provider %s" % name,
text="Provider with name %s already known. Do you want to "
"use it for this session?"
% name,
default=True
):
return known_providers_by_name[name]
elif path.lexists(filename):
ui.error(
yield ses, cred
except gh.BadCredentialsException as exc:
lgr.error("Bad Github credentials")
except (gh.TwoFactorException, gh.GithubException) as exc:
# With github 1.43.5, in comparison to 1.40 we get a "regular"
# GithubException for some reason, yet to check/report upstream
# so we will just check for the expected in such cases messages
if not (
isinstance(exc, gh.GithubException) and
getattr(exc, 'data', {}).get('message', '').startswith(
'Must specify two-factor authentication OTP code')
):
raise
# 2FA - we need to interact!
if not ui.is_interactive:
# Or should we just allow to pass
raise RuntimeError(
"Cannot proceed with 2FA for Github - UI is not interactive. "
"Please 'manually' establish token based authentication "
"with Github and specify it in %s config"
% CONFIG_HUB_TOKEN_FIELD
)
if not ui.yesno(
title="GitHub credentials - %s uses 2FA" % user_name,
text="Generate a GitHub token to proceed? "
"If you already have a token for the account, "
"just say 'no' now and specify it in config (%s), "
"otherwise say 'yes' "
% (CONFIG_HUB_TOKEN_FIELD,)
):
return
# but I couldn't seem to get eval_results to detect the filter when I
# used
#
# result_renderer = "json_pp"
# result_filter = lambda x: ...
#
# Also, I want to keep the "message" key for errors.
from datalad.ui import ui
to_render = {}
if res["status"] == "error":
to_render = dict(res.items())
elif "report" in res and res["report"]["objects"]:
to_render = {k: v for k, v in res.items()
if k not in ["status", "message", "logger"]}
if to_render:
ui.message(json.dumps(to_render, sort_keys=True, indent=2))