def setup(mute_stdout=False):
# logging.basicConfig()
if mute_stdout:
handler = logging.NullHandler()
else:
formatter_str = '%(asctime)s %(levelname)s %(message)s'
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(formatter_str))
# Add handler to logger
logger = logging.getLogger(_product_name)
logger.addHandler(handler)
# disable unnecessary information capture
logging.logThreads = 0
logging.logProcesses = 0
# to make sure each log record does not have a source file name attached
# pylint: disable=protected-access
logging._srcfile = None
# pylint: enable=protected-access
python类NullHandler()的实例源码
def _setup_logging(logger, options):
formatter = logging.Formatter('%(asctime)s %(process)s %(levelname)s '
'%(message)s')
logger.setLevel(logging.DEBUG)
if options["log_file"]:
filename = options["log_file"]
file_handler = logging.handlers.WatchedFileHandler(filename)
file_handler.setFormatter(formatter)
if options["debug"]:
file_handler.setLevel(logging.DEBUG)
else:
file_handler.setLevel(logging.INFO)
logger.addHandler(file_handler)
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.CRITICAL)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
if options["sentry_dsn"] and _has_raven:
client = raven.Client(options["sentry_dsn"],
enable_breadcrumbs=False,
transport=raven.transport.HTTPTransport)
# Add Sentry handle to application logger.
sentry_handler = SentryHandler(client)
sentry_handler.setLevel(logging.WARNING)
logger.addHandler(sentry_handler)
null_loggers = [
logging.getLogger("sentry.errors"),
logging.getLogger("sentry.errors.uncaught")
]
for null_logger in null_loggers:
null_logger.handlers = [logging.NullHandler()]
if options["debug"]:
stream_handler.setLevel(logging.DEBUG)
elif options["info"]:
stream_handler.setLevel(logging.INFO)
def __init__(self, *flags, **options):
self._flags = flagNone
if options.get('printer') is not None:
self._printer = options.get('printer')
elif self.defaultPrinter is not None:
self._printer = self.defaultPrinter
if 'loggerName' in options:
# route our logs to parent logger
self._printer = Printer(
logger=logging.getLogger(options['loggerName']),
handler=NullHandler()
)
else:
self._printer = Printer()
self('running pyasn1 version %s' % __version__)
for f in flags:
inverse = f and f[0] in ('!', '~')
if inverse:
f = f[1:]
try:
if inverse:
self._flags &= ~flagMap[f]
else:
self._flags |= flagMap[f]
except KeyError:
raise error.PyAsn1Error('bad debug flag %s' % f)
self('debug category \'%s\' %s' % (f, inverse and 'disabled' or 'enabled'))
def __init__(self, *flags, **options):
self._flags = flagNone
if options.get('printer') is not None:
self._printer = options.get('printer')
elif self.defaultPrinter is not None:
self._printer = self.defaultPrinter
if 'loggerName' in options:
# route our logs to parent logger
self._printer = Printer(
logger=logging.getLogger(options['loggerName']),
handler=NullHandler()
)
else:
self._printer = Printer()
self('running pyasn1 version %s' % __version__)
for f in flags:
inverse = f and f[0] in ('!', '~')
if inverse:
f = f[1:]
try:
if inverse:
self._flags &= ~flagMap[f]
else:
self._flags |= flagMap[f]
except KeyError:
raise error.PyAsn1Error('bad debug flag %s' % f)
self('debug category \'%s\' %s' % (f, inverse and 'disabled' or 'enabled'))
def test_inner_logger_build(self):
plugin = self.loader.factory('sc-plugins:InnerLogger', logger={'type': 'logging:Logger',
'params': {'name': 'foo.bar.test.1',
'handlers': ['logging:NullHandler']}})
self.assertIsInstance(plugin, InnerLogger)
self.assertEqual(plugin.logger, getLogger('foo.bar.test.1'))
self.assertEqual(len(plugin.logger.handlers), 1)
self.assertIsInstance(plugin.logger.handlers[0], NullHandler)
def configure_test_logging():
"""Add a NullHandler to root logger.
Prevents unconfigured logging from erroring, and swallows all logging,
which is usually what you want for unit tests. Unit test fixtures can
still add their own loggers to assert against log messages if needed.
"""
set_logger_class()
handler = logging.NullHandler()
add_talisker_handler(logging.NOTSET, handler)
configure_warnings(True)
def __init__(self, sources, targets, chunk_size=1000):
"""
Initialize the builder the framework.
Args:
sources([Store]): list of source stores
targets([Store]): list of target stores
chunk_size(int): chunk size for processing
"""
self.sources = sources
self.targets = targets
self.chunk_size = chunk_size
self.logger = logging.getLogger(type(self).__name__)
self.logger.addHandler(logging.NullHandler())
def __init__(self, builders):
"""
Initialize with a list of builders
Args:
builders(list): list of builders
"""
self.builders = builders
self.logger = logging.getLogger(type(self).__name__)
self.logger.addHandler(logging.NullHandler())
def __init__(self, *flags, **options):
self._flags = flagNone
if 'loggerName' in options:
# route our logs to parent logger
self._printer = Printer(
logger=logging.getLogger(options['loggerName']),
handler=NullHandler()
)
elif 'printer' in options:
self._printer = options.get('printer')
else:
self._printer = self.defaultPrinter
self._printer('running pyasn1 %s, debug flags %s' % (__version__, ', '.join(flags)))
for flag in flags:
inverse = flag and flag[0] in ('!', '~')
if inverse:
flag = flag[1:]
try:
if inverse:
self._flags &= ~flagMap[flag]
else:
self._flags |= flagMap[flag]
except KeyError:
raise error.PyAsn1Error('bad debug flag %s' % flag)
self._printer("debug category '%s' %s" % (flag, inverse and 'disabled' or 'enabled'))
def __init__(self, jabberid, password, bosh_service, hold=1, wait=60,
logger=None):
"""
:param jabberid:
:param password:
:param bosh_service:
:param hold:
:param wait:
:param logger:
:return:
"""
if logger is None:
logger = logging.getLogger('bosh_client')
logger.addHandler(logging.NullHandler())
logger.debug("BOSHClient jabberid:%s; password:%s; bosh_service:%s",
jabberid, password, bosh_service)
self.rid = random.randint(0, 10000000)
self.jabberid = JID(jabberid)
self.password = password
self.authid = None
self.sid = None
self.logged_in = False
self.headers = {
"User-Agent": "do-portal-bosh-client/0.1",
"Content-type": "text/xml",
"Accept": "text/xml"
}
self.bosh_service = urlparse(bosh_service)
self.request_session = requests.Session()
self.request_session.headers.update(self.headers)
self.logged_in = self.start_session_and_auth(hold, wait)
# Close it
self.request_session.close()
logger.debug("BOSH Logged In: %s", self.logged_in)
return
def __init__(self, *flags, **options):
self._flags = flagNone
if options.get('printer') is not None:
self._printer = options.get('printer')
elif self.defaultPrinter is not None:
self._printer = self.defaultPrinter
if 'loggerName' in options:
# route our logs to parent logger
self._printer = Printer(
logger=logging.getLogger(options['loggerName']),
handler=NullHandler()
)
else:
self._printer = Printer()
self('running pyasn1 version %s' % __version__)
for f in flags:
inverse = f and f[0] in ('!', '~')
if inverse:
f = f[1:]
try:
if inverse:
self._flags &= ~flagMap[f]
else:
self._flags |= flagMap[f]
except KeyError:
raise error.PyAsn1Error('bad debug flag %s' % f)
self('debug category \'%s\' %s' % (f, inverse and 'disabled' or 'enabled'))
def get_default_logger():
"""Get the default dkimpy logger."""
logger = logging.getLogger('dkimpy')
if not logger.handlers:
logger.addHandler(NullHandler())
return logger
def _setup_logging():
"""Setup logging to log to nowhere by default.
For details, see:
http://docs.python.org/3/howto/logging.html#library-config
Internal function.
"""
import logging
logger = logging.getLogger('spotify-connect')
handler = logging.NullHandler()
logger.addHandler(handler)
def suppress_logging() -> None:
"""Suppress log output to stdout.
This function is intended to be used in test's setup. This function removes
log handler of ``wdom`` logger and set NullHandler to suppress log.
"""
from wdom import options
options.root_logger.removeHandler(options._log_handler)
options.root_logger.addHandler(logging.NullHandler())
def __init__(self, *flags, **options):
self._flags = flagNone
if options.get('printer') is not None:
self._printer = options.get('printer')
elif self.defaultPrinter is not None:
self._printer = self.defaultPrinter
if 'loggerName' in options:
# route our logs to parent logger
self._printer = Printer(
logger=logging.getLogger(options['loggerName']),
handler=NullHandler()
)
else:
self._printer = Printer()
self('running pyasn1 version %s' % __version__)
for f in flags:
inverse = f and f[0] in ('!', '~')
if inverse:
f = f[1:]
try:
if inverse:
self._flags &= ~flagMap[f]
else:
self._flags |= flagMap[f]
except KeyError:
raise error.PyAsn1Error('bad debug flag %s' % f)
self('debug category \'%s\' %s' % (f, inverse and 'disabled' or 'enabled'))
def _setup_logging_from_flags():
ops_root = getLogger().logger
for handler in ops_root.handlers:
ops_root.removeHandler(handler)
logpath = _get_log_file_path()
if logpath:
filelog = logging.handlers.WatchedFileHandler(logpath)
ops_root.addHandler(filelog)
mode = int(options.logfile_mode, 8)
st = os.stat(logpath)
if st.st_mode != (stat.S_IFREG | mode):
os.chmod(logpath, mode)
for handler in ops_root.handlers:
handler.setFormatter(logging.Formatter(fmt=options.log_format,
datefmt=options.log_date_format))
if options.verbose or options.debug:
ops_root.setLevel(logging.DEBUG)
else:
ops_root.setLevel(logging.INFO)
root = logging.getLogger()
for handler in root.handlers:
root.removeHandler(handler)
handler = NullHandler()
handler.setFormatter(logging.Formatter())
root.addHandler(handler)
def configure_logger(
logger=None,
log_level='DEBUG',
no_log=False,
file_log=False,
console_log=True,
log_file=None):
if not logger:
logger = get_logger()
if no_log:
logger.setLevel(logging.ERROR)
logger.addHandler(logging.NullHandler())
else:
logger.setLevel(log_level.upper())
fmt = (
"%(asctime)s - %(message)s"
)
fmtr = formatter()
if console_log:
if USE_COLORED_LOGS:
coloredlogs.install(level=os.environ.get('COLOREDLOGS_LOG_LEVEL', log_level.upper()),
fmt=fmt,
field_styles=FIELD_STYLES,
level_styles=LEVEL_STYLES,
overridefmt=LEVEL_FORMATS)
else:
sh = logging.StreamHandler()
sh.setFormatter(fmtr)
sh.setLevel(log_level.upper())
logger.addHandler(sh)
if file_log:
if log_file is not None:
func_log = os.path.abspath(log_file)
os.mkdir(os.path.dirname(func_log))
fh = logging.FileHandler(func_log)
fh.setFormatter(fmtr)
fh.setLevel(log_level)
logger.addHandler(fh)
def reset_logger(log_comp):
"""
Reset the logger for the specified log component (unless it is the root
logger) to add a NullHandler if it does not have any handlers. Having a
handler prevents a log request to be propagated to the parent logger.
"""
name = LOGGER_NAMES[log_comp]
logger = logging.getLogger(name)
if name != '' and not logger.handlers:
logger.addHandler(logging.NullHandler())
def get_logger(name):
"""
Return a :class:`~py:logging.Logger` object with the specified name.
A :class:`~py:logging.NullHandler` handler is added to the logger if it
does not have any handlers yet and if it is not the Python root logger.
This prevents the propagation of log requests up the Python logger
hierarchy, and therefore causes this package to be silent by default.
"""
logger = logging.getLogger(name)
if name != '' and not logger.handlers:
logger.addHandler(logging.NullHandler())
return logger
def log_msg(string):
logging.getLogger(__name__).addHandler(NullHandler())
logging.info(string)
