import logging
import requests
import socket
import sys
from builtins import str
from datetime import datetime
import click
import pkg_resources
from six import PY3, BytesIO, StringIO, string_types
from dtale.utils import make_list
logger = logging.getLogger(__name__)
LOG_LEVELS = dict(
debug=logging.DEBUG,
info=logging.INFO,
warning=logging.WARNING,
error=logging.ERROR,
critical=logging.CRITICAL,
)
[docs]def setup_logging(logfile, log_level, verbose=False):
"""
Utility method for setting up logging configuration
:param logfile: location of D-Tale logs
:type logfile: str
:param log_level: D-Tale's implementation of standard logging levels
:type log_level: str, options are debug, info, warning, error or, critical
:param verbose: turns on verbose logging, defaults to False
:type verbose: bool, optional
:return:
"""
if log_level == "verbose" or verbose:
log_level = LOG_LEVELS["debug"]
elif log_level:
log_level = LOG_LEVELS[log_level]
else:
log_level = LOG_LEVELS["info"]
logging.getLogger().handlers = []
fmt = "%(asctime)s - %(levelname)-8s - %(message)s"
try:
logging.basicConfig(format=fmt, level=log_level)
except BaseException:
# #202: when using Pyzo IDE the basicConfig has already been set and if you try to set it again
# then it will cause a maximum recursion exception
logging.getLogger().setLevel(log_level)
if logfile:
fh = logging.FileHandler(logfile, mode="w")
fh.setFormatter(logging.Formatter(fmt))
logging.getLogger().addHandler(fh)
for handler in logging.getLogger().handlers:
handler.setLevel(log_level)
handler.setFormatter(logging.Formatter(fmt))
logger.debug("{}".format(" ".join(sys.argv)))
try:
logger.debug("Hostname: {}".format(socket.gethostname()))
except Exception as e:
logger.exception(e)
[docs]def loader_options(key, params):
"""
Builds a decorator which will append a list of click parameters based on
a combination of prefix & parameter name
loader_options('foo', ['bar', 'baz']) => click.option('--foo-bar'), click.option('--foo-baz')
:param key: parameter prefix
:type key: str
:param params: parameters names
:type params: list
:return: filter
:rtype: func
"""
def decorator(f):
for p in params:
f = click.option(
"--" + key + "-" + p, help="Override {} {}".format(key, p)
)(f)
return f
return decorator
[docs]def get_loader_options(key, properties, options):
"""
Filters dictionary of click parameters for ones which start with a certain prefix
:param key: click option prefix
:type key: str
:param properties: sub-properties of the click command key
:type properties: list of str
:param options: click options
:type options: dict
:return: dictionary of click options with start with key
:rtype: dict
"""
def _build_key(option):
segs = option.split("_")
if len(segs) == 1:
return ""
return option.split("{}_".format(key))[-1]
final_key = "_".join(key.split("-"))
selected_opts = (
[final_key]
if not len(make_list(properties))
else [
"{}_{}".format(final_key, prop["name"] if isinstance(prop, dict) else prop)
for prop in properties
]
)
return dict(
(
(_build_key(k), v)
for k, v in options.items()
if k in selected_opts
if v is not None
)
)
[docs]def get_log_options(options):
"""
Click logging options (logfile, log_level, verbose)
"""
names = ["logfile", "log_level", "verbose"]
return get_named_options(names, options)
[docs]def get_named_options(names, options):
"""
Filters click options for options in a specific subset of names
"""
return {k: v for k, v in options.items() if k in names}
[docs]def get_args(click_wrapper):
"""
Retrieves arguments being passed in from the command-line and applying them to a click command
:param click_wrapper: click context
:return: list of arguments
:rtype: list
"""
# Get the options defined on cli
params = []
for opt in click_wrapper.params:
params.extend(opt.opts)
args = sys.argv[1:]
# Reorder the args so that cli options come before
# subcommand options. This is a click limitation.
head, tail = [], []
for a in args:
if a in params or head and head[-1] in params:
head.append(a)
else:
tail.append(a)
return head + tail
[docs]def run(click_wrapper):
"""
Wrapper function for executing click commands and handling exceptions produced from
click commands with versioning boilerplate
:param click_wrapper: click context
"""
t1 = datetime.now()
try:
try:
dtale_name = "dtale"
bld_info, ver_info = retrieve_meta_info_and_version(dtale_name)
logger.debug("{} bld: {}".format(dtale_name, bld_info))
logger.debug("{} ver: {}".format(dtale_name, ver_info))
except BaseException:
logger.debug("failure to retrieve metadata for: {}".format(dtale_name))
args = get_args(click_wrapper)
click_wrapper(args)
sys.exit(0)
except Exception as ex:
logger.exception("Fatal error: " + str(ex))
sys.exit(1)
finally:
logger.info("Elapsed time: %s" % (datetime.now() - t1))
[docs]def loader_prop_keys(prop_cfgs):
return [p if isinstance(p, string_types) else p["name"] for p in prop_cfgs]
[docs]def handle_str_content(resp):
return BytesIO(resp.content) if PY3 else StringIO(resp.content.decode("utf-8"))
[docs]def is_url(path):
return path.startswith("http://") or path.startswith("https://")
[docs]def handle_path(path, kwargs, resp_handler=handle_str_content):
if is_url(path):
proxy = kwargs.pop("proxy", None)
req_kwargs = {}
if proxy is not None:
req_kwargs["proxies"] = dict(http=proxy, https=proxy)
resp = requests.get(path, **req_kwargs)
assert resp.status_code == 200
return resp_handler(resp)
return path