def _script_magics_default(self):
"""default to a common list of programs"""
defaults = [
'sh',
'bash',
'perl',
'ruby',
'python',
'python2',
'python3',
'pypy',
]
if os.name == 'nt':
defaults.extend([
'cmd',
])
return defaults
python类default()的实例源码
def _script_magics_default(self):
"""default to a common list of programs"""
defaults = [
'sh',
'bash',
'perl',
'ruby',
'python',
'python2',
'python3',
'pypy',
]
if os.name == 'nt':
defaults.extend([
'cmd',
])
return defaults
def get_default_editor():
try:
ed = os.environ['EDITOR']
if not PY3:
ed = ed.decode()
return ed
except KeyError:
pass
except UnicodeError:
warn("$EDITOR environment variable is not pure ASCII. Using platform "
"default editor.")
if os.name == 'posix':
return 'vi' # the only one guaranteed to be there!
else:
return 'notepad' # same in Windows!
# conservatively check for tty
# overridden streams can result in things like:
# - sys.stdin = None
# - no isatty method
def user_for_token(self, token, use_cache=True):
"""Ask the Hub to identify the user for a given token.
Args:
token (str): the token
use_cache (bool): Specify use_cache=False to skip cached cookie values (default: True)
Returns:
user_model (dict): The user model, if a user is identified, None if authentication fails.
The 'name' field contains the user's name.
"""
return self._check_hub_authorization(
url=url_path_join(self.api_url,
"authorizations/token",
quote(token, safe='')),
cache_key='token:%s' % token,
use_cache=use_cache,
)
def _update_core_values(name=None):
def _func(self, change=None):
core = get_proxy_or_core()
if name is None:
core.num_threads = self.core_num_threads
core.add_cache = self.core_add_cache
core.accept_lowercase = self.core_accept_lowercase
# There is no obvious default for max_cache_size
if self.core_max_cache_size is not None:
core.max_cache_size = self.core_max_cache_size
else:
setattr(core, name, change.new)
return _func
def get_default_editor():
try:
ed = os.environ['EDITOR']
if not py3compat.PY3:
ed = ed.decode()
return ed
except KeyError:
pass
except UnicodeError:
warn("$EDITOR environment variable is not pure ASCII. Using platform "
"default editor.")
if os.name == 'posix':
return 'vi' # the only one guaranteed to be there!
else:
return 'notepad' # same in Windows!
def mainloop(self, display_banner=None):
"""Start the mainloop.
If an optional banner argument is given, it will override the
internally created default banner.
"""
with self.builtin_trap, self.display_trap:
while 1:
try:
self.interact(display_banner=display_banner)
#self.interact_with_readline()
# XXX for testing of a readline-decoupled repl loop, call
# interact_with_readline above
break
except KeyboardInterrupt:
# this should not be necessary, but KeyboardInterrupt
# handling seems rather unpredictable...
self.write("\nKeyboardInterrupt in interact()\n")
def _script_magics_default(self):
"""default to a common list of programs"""
defaults = [
'sh',
'bash',
'perl',
'ruby',
'python',
'python2',
'python3',
'pypy',
]
if os.name == 'nt':
defaults.extend([
'cmd',
])
return defaults
def get_default_editor():
try:
ed = os.environ['EDITOR']
if not PY3:
ed = ed.decode()
return ed
except KeyError:
pass
except UnicodeError:
warn("$EDITOR environment variable is not pure ASCII. Using platform "
"default editor.")
if os.name == 'posix':
return 'vi' # the only one guaranteed to be there!
else:
return 'notepad' # same in Windows!
# conservatively check for tty
# overridden streams can result in things like:
# - sys.stdin = None
# - no isatty method
def _script_magics_default(self):
"""default to a common list of programs"""
defaults = [
'sh',
'bash',
'perl',
'ruby',
'python',
'python2',
'python3',
'pypy',
]
if os.name == 'nt':
defaults.extend([
'cmd',
])
return defaults
def _get_bucket(self, name, throw=False):
"""
Get the bucket by it's name. Uses cache by default.
:param name: bucket name.
:param throw: If True raises NotFound exception, otherwise, returns
None.
:return: instance of :class:`google.cloud.storage.Bucket` or None.
"""
if not self.cache_buckets:
try:
return self.client.get_bucket(name)
except NotFound:
if throw:
raise
return None
try:
cache = self._bucket_cache
except AttributeError:
self._bucket_cache = cache = {}
try:
return cache[name]
except KeyError:
try:
bucket = self.client.get_bucket(name)
except BrokenPipeError as e:
if e.errno in (None, errno.EPIPE):
return self._get_bucket(name, throw)
else:
raise
except (BadRequest, NotFound):
if throw:
raise
return None
cache[name] = bucket
return bucket
def enable_win_unicode_console(self):
if sys.version_info >= (3, 6):
# Since PEP 528, Python uses the unicode APIs for the Windows
# console by default, so WUC shouldn't be needed.
return
import win_unicode_console
if PY3:
win_unicode_console.enable()
else:
# https://github.com/ipython/ipython/issues/9768
from win_unicode_console.streams import (TextStreamWrapper,
stdout_text_transcoded, stderr_text_transcoded)
class LenientStrStreamWrapper(TextStreamWrapper):
def write(self, s):
if isinstance(s, bytes):
s = s.decode(self.encoding, 'replace')
self.base.write(s)
stdout_text_str = LenientStrStreamWrapper(stdout_text_transcoded)
stderr_text_str = LenientStrStreamWrapper(stderr_text_transcoded)
win_unicode_console.enable(stdout=stdout_text_str,
stderr=stderr_text_str)
enterprisegatewayapp.py 文件源码
项目:enterprise_gateway
作者: jupyter-incubator
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def _default_log_format(self):
"""override default log format to include milliseconds"""
return u"%(color)s[%(levelname)1.1s %(asctime)s.%(msecs).03d %(name)s]%(end_color)s %(message)s"
# Impersonation enabled
def _check_hub_authorization(self, url, cache_key=None, use_cache=True):
"""Identify a user with the Hub
Args:
url (str): The API URL to check the Hub for authorization
(e.g. http://127.0.0.1:8081/hub/api/authorizations/token/abc-def)
cache_key (str): The key for checking the cache
use_cache (bool): Specify use_cache=False to skip cached cookie values (default: True)
Returns:
user_model (dict): The user model, if a user is identified, None if authentication fails.
Raises an HTTPError if the request failed for a reason other than no such user.
"""
if use_cache:
if cache_key is None:
raise ValueError("cache_key is required when using cache")
# check for a cached reply, so we don't check with the Hub if we don't have to
cached = self.cache.get(cache_key)
if cached is not None:
return cached
try:
r = requests.get(url,
headers = {
'Authorization' : 'token %s' % self.api_token,
},
)
except requests.ConnectionError:
msg = "Failed to connect to Hub API at %r." % self.api_url
msg += " Is the Hub accessible at this URL (from host: %s)?" % socket.gethostname()
if '127.0.0.1' in self.api_url:
msg += " Make sure to set c.JupyterHub.hub_ip to an IP accessible to" + \
" single-user servers if the servers are not on the same host as the Hub."
raise HTTPError(500, msg)
data = None
if r.status_code == 404:
app_log.warning("No Hub user identified for request")
elif r.status_code == 403:
app_log.error("I don't have permission to check authorization with JupyterHub, my auth token may have expired: [%i] %s", r.status_code, r.reason)
raise HTTPError(500, "Permission failure checking authorization, I may need a new token")
elif r.status_code >= 500:
app_log.error("Upstream failure verifying auth token: [%i] %s", r.status_code, r.reason)
raise HTTPError(502, "Failed to check authorization (upstream problem)")
elif r.status_code >= 400:
app_log.warning("Failed to check authorization: [%i] %s", r.status_code, r.reason)
raise HTTPError(500, "Failed to check authorization")
else:
data = r.json()
app_log.debug("Received request from Hub user %s", data)
if use_cache:
# cache result
self.cache[cache_key] = data
return data
def _make_style_from_name(self, name):
"""
Small wrapper that make an IPython compatible style from a style name
We need that to add style for prompt ... etc.
"""
style_overrides = {}
if name == 'legacy':
legacy = self.colors.lower()
if legacy == 'linux':
style_cls = get_style_by_name('monokai')
style_overrides = _style_overrides_linux
elif legacy == 'lightbg':
style_overrides = _style_overrides_light_bg
style_cls = get_style_by_name('pastie')
elif legacy == 'neutral':
# The default theme needs to be visible on both a dark background
# and a light background, because we can't tell what the terminal
# looks like. These tweaks to the default theme help with that.
style_cls = get_style_by_name('default')
style_overrides.update({
Token.Number: '#007700',
Token.Operator: 'noinherit',
Token.String: '#BB6622',
Token.Name.Function: '#2080D0',
Token.Name.Class: 'bold #2080D0',
Token.Name.Namespace: 'bold #2080D0',
Token.Prompt: '#009900',
Token.PromptNum: '#00ff00 bold',
Token.OutPrompt: '#990000',
Token.OutPromptNum: '#ff0000 bold',
})
elif legacy =='nocolor':
style_cls=_NoStyle
style_overrides = {}
else :
raise ValueError('Got unknown colors: ', legacy)
else :
style_cls = get_style_by_name(name)
style_overrides = {
Token.Prompt: '#009900',
Token.PromptNum: '#00ff00 bold',
Token.OutPrompt: '#990000',
Token.OutPromptNum: '#ff0000 bold',
}
style_overrides.update(self.highlighting_style_overrides)
style = PygmentsStyle.from_defaults(pygments_style_cls=style_cls,
style_dict=style_overrides)
return style