def initialize_cli(options):
from ._logger import set_log_level
debug_format_str = (
"[{record.level_name}] {record.channel} {record.func_name} "
"({record.lineno}): {record.message}")
if options.log_level == logbook.DEBUG:
info_format_str = debug_format_str
else:
info_format_str = (
"[{record.level_name}] {record.channel}: {record.message}")
logbook.StderrHandler(
level=logbook.DEBUG, format_string=debug_format_str
).push_application()
logbook.StderrHandler(
level=logbook.INFO, format_string=info_format_str
).push_application()
set_log_level(options.log_level)
spr.SubprocessRunner.is_save_history = True
if options.is_output_stacktrace:
spr.SubprocessRunner.is_output_stacktrace = (
options.is_output_stacktrace)
python类INFO的实例源码
def test_stderr(
self, capsys, command, ignore_stderr_regexp, out_regexp, expected):
import logbook
import subprocrunner
logbook.StderrHandler(
level=logbook.DEBUG).push_application()
subprocrunner.set_log_level(logbook.INFO)
runner = SubprocessRunner(
command, ignore_stderr_regexp=ignore_stderr_regexp)
runner.run()
assert is_null_string(runner.stdout.strip())
assert is_not_null_string(runner.stderr.strip())
out, err = capsys.readouterr()
print("[sys stdout]\n{}\n".format(out))
print("[sys stderr]\n{}\n".format(err))
print("[proc stdout]\n{}\n".format(runner.stdout))
print("[proc stderr]\n{}\n".format(runner.stderr))
actual = out_regexp.search(err) is not None
assert actual == expected
def initialize_log_handler(log_level):
debug_format_str = (
"[{record.level_name}] {record.channel} {record.func_name} "
"({record.lineno}): {record.message}")
if log_level == logbook.DEBUG:
info_format_str = debug_format_str
else:
info_format_str = (
"[{record.level_name}] {record.channel}: {record.message}")
logbook.StderrHandler(
level=logbook.DEBUG, format_string=debug_format_str
).push_application()
logbook.StderrHandler(
level=logbook.INFO, format_string=info_format_str
).push_application()
def __init__(self,
log_level=LogLevel.INFO,
format_str='[{record.time:%Y-%m-%d %H:%M:%S}] - {record.channel} - {record.level_name} '
'- {record.message}'):
self.logger = Logger('WindAdapter')
set_datetime_format('local')
StreamHandler(sys.stdout, format_string=format_str).push_application()
FileHandler('WindAdapter.log', bubble=True, format_string=format_str).push_application()
self.set_level(log_level)
def set_level(self, log_level):
if log_level.lower() == LogLevel.INFO:
self.logger.level = logbook.INFO
elif log_level.lower() == LogLevel.WARNING:
self.logger.level = logbook.WARNING
elif log_level.lower() == LogLevel.CRITICAL:
self.logger.level = logbook.CRITICAL
elif log_level.lower() == LogLevel.NOTSET:
self.logger.level = logbook.NOTSET
def __get_logbook_logging_level(level_str):
# logbook levels:
# CRITICAL = 15
# ERROR = 14
# WARNING = 13
# NOTICE = 12
# INFO = 11
# DEBUG = 10
# TRACE = 9
# NOTSET = 0
level_str = level_str.upper().strip()
if level_str == 'CRITICAL':
return logbook.CRITICAL
elif level_str == 'ERROR':
return logbook.ERROR
elif level_str == 'WARNING':
return logbook.WARNING
elif level_str == 'NOTICE':
return logbook.NOTICE
elif level_str == 'INFO':
return logbook.INFO
elif level_str == 'DEBUG':
return logbook.DEBUG
elif level_str == 'TRACE':
return logbook.TRACE
elif level_str == 'NOTSET':
return logbook.NOTSET
else:
raise ValueError("Unknown logbook log level: {}".format(level_str))
def cmd(ctx, is_append_table, index_list, verbosity_level, log_level):
ctx.obj[Context.IS_APPEND_TABLE] = is_append_table
ctx.obj[Context.INDEX_LIST] = index_list.split(",")
ctx.obj[Context.VERBOSITY_LEVEL] = verbosity_level
ctx.obj[Context.LOG_LEVEL] = (
logbook.INFO if log_level is None else log_level)
def _add_log_level_argument_group(self):
dest = "log_level"
group = self.parser.add_mutually_exclusive_group()
group.add_argument(
"--debug", dest=dest, action="store_const",
const=logbook.DEBUG, default=logbook.INFO,
help="for debug print.")
group.add_argument(
"--quiet", dest=dest, action="store_const",
const=logbook.NOTSET, default=logbook.INFO,
help="suppress execution log messages.")
return group
def __get_logging_method(log_level):
method_table = {
logbook.DEBUG: logger.debug,
logbook.INFO: logger.info,
logbook.WARNING: logger.warning,
logbook.ERROR: logger.error,
logbook.CRITICAL: logger.critical,
}
method = method_table.get(log_level)
if method is None:
raise ValueError("unknown log level: {}".format(log_level))
return method
log_service.py 文件源码
项目:cookiecutter-pyramid-talk-python-starter
作者: mikeckennedy
项目源码
文件源码
阅读 26
收藏 0
点赞 0
评论 0
def __get_logbook_logging_level(level_str):
# logbook levels:
# CRITICAL = 15
# ERROR = 14
# WARNING = 13
# NOTICE = 12
# INFO = 11
# DEBUG = 10
# TRACE = 9
# NOTSET = 0
level_str = level_str.upper().strip()
if level_str == 'CRITICAL':
return logbook.CRITICAL
elif level_str == 'ERROR':
return logbook.ERROR
elif level_str == 'WARNING':
return logbook.WARNING
elif level_str == 'NOTICE':
return logbook.NOTICE
elif level_str == 'INFO':
return logbook.INFO
elif level_str == 'DEBUG':
return logbook.DEBUG
elif level_str == 'TRACE':
return logbook.TRACE
elif level_str == 'NOTSET':
return logbook.NOTSET
else:
raise ValueError("Unknown logbook log level: {}".format(level_str))
def __get_logbook_logging_level(level_str):
# logbook levels:
# CRITICAL = 15
# ERROR = 14
# WARNING = 13
# NOTICE = 12
# INFO = 11
# DEBUG = 10
# TRACE = 9
# NOTSET = 0
level_str = level_str.upper().strip()
if level_str == 'CRITICAL':
return logbook.CRITICAL
elif level_str == 'ERROR':
return logbook.ERROR
elif level_str == 'WARNING':
return logbook.WARNING
elif level_str == 'NOTICE':
return logbook.NOTICE
elif level_str == 'INFO':
return logbook.INFO
elif level_str == 'DEBUG':
return logbook.DEBUG
elif level_str == 'TRACE':
return logbook.TRACE
elif level_str == 'NOTSET':
return logbook.NOTSET
else:
raise ValueError("Unknown logbook log level: {}".format(level_str))
def __init__(self, *args, **kwargs):
super().__init__(command_prefix=when_mentioned_or(setup_file["discord"]["command_prefix"]),
description="A bot for weebs programmed by Recchan")
# Set a custom user agent for Pixie
self.http.user_agent = user_agent
# Logging setup
redirect_logging()
StreamHandler(sys.stderr).push_application()
self.logger = Logger("Pixie")
self.logger.level = getattr(logbook, setup_file.get("log_level", "INFO"), logbook.INFO)
logging.root.setLevel(self.logger.level)
def _get_logging_level(verbosity):
# noinspection PyPackageRequirements
import logbook
return {
1: logbook.CRITICAL,
2: logbook.ERROR,
3: logbook.WARNING,
4: logbook.NOTICE,
5: logbook.INFO,
6: logbook.DEBUG,
7: logbook.TRACE,
}[verbosity]
def main():
options = parse_option()
initialize_cli(options)
if is_execute_tc_command(options.tc_command_output):
check_tc_command_installation()
is_delete_all = options.is_delete_all
else:
subprocrunner.SubprocessRunner.default_is_dry_run = True
is_delete_all = True
set_logger(False)
try:
verify_network_interface(options.device)
except NetworkInterfaceNotFoundError as e:
logger.error("{:s}: {}".format(e.__class__.__name__, e))
return errno.EINVAL
subprocrunner.SubprocessRunner.clear_history()
tc = create_tc_obj(options)
if options.log_level == logbook.INFO:
subprocrunner.set_log_level(logbook.ERROR)
normalize_tc_value(tc)
return_code = 0
if is_delete_all:
return_code = tc.delete_all_tc()
else:
return_code = tc.delete_tc()
command_history = "\n".join(tc.get_command_history())
if options.tc_command_output == TcCommandOutput.STDOUT:
print(command_history)
return return_code
elif options.tc_command_output == TcCommandOutput.SCRIPT:
set_logger(True)
write_tc_script(
Tc.Command.TCDEL, command_history, filename_suffix=options.device)
return return_code
logger.debug("command history\n{}".format(command_history))
return return_code
def parse_option():
parser = argparse.ArgumentParser()
if is_use_stdin():
parser.add_argument(
"destination_or_file", nargs="+",
help="")
parser.add_argument(
"--max-workers", type=int,
help="""a number of threads for when multiple destination/file
specified. defaults to equals to two times number of cores.
""")
parser.add_argument(
"--indent", type=int, default=4,
help="""JSON output will be pretty-printed with the indent level.
(default= %(default)s)
""")
loglevel_dest = "log_level"
group = parser.add_mutually_exclusive_group()
group.add_argument(
"--debug", dest=loglevel_dest, action="store_const",
const=logbook.DEBUG, default=logbook.INFO,
help="for debug print.")
group.add_argument(
"--quiet", dest=loglevel_dest, action="store_const",
const=logbook.NOTSET, default=logbook.INFO,
help="suppress execution log messages.")
group = parser.add_argument_group("Ping Options")
group.add_argument(
"-c", "--count", type=int,
help="""stop after sending the count.
see also ping(8) [-c count] option description.
""")
group.add_argument(
"-w", "--deadline", type=float,
help="""timeout in seconds.
see also ping(8) [-w deadline] option description.
""")
group.add_argument(
"-I", "--interface", dest="interface",
help="network interface")
return parser.parse_args()