Source code for quicklogging.stream_wrapper
"""
Module quicklogging.stream_wrapper
====================================
.. autoclass:: StreamWrapper
"""
import warnings
from .base import get_logger
[docs]class StreamWrapper:
"""Implementation detail"""
INSTANCES = {}
def __init__(self, original_stream, logfunc, catch_all=False, warn=False):
self.original_stream = original_stream
self.logfunc = logfunc
self.catch_all = catch_all
self.warn = warn
self.fed_loggers = set()
self.fed_logger_children = set()
self.isatty = self.original_stream.isatty
def _intercepting_logger(self):
target_logger = get_logger(stackoverhead=2)
if self.catch_all:
return target_logger
target_logger_name = target_logger.name
if (
target_logger_name in self.fed_loggers
) or any(
target_logger_name.startswith(parent)
for parent in self.fed_logger_children
):
return target_logger
return None
def flush(self):
pass
def write(self, message):
message = message.strip()
if not message:
return
if self.warn:
warnings.warn("print() detected: {}".format(message))
target_logger = self._intercepting_logger()
if target_logger is None:
self.original_stream.write(message)
return
self.logfunc(message, stackoverhead=2)
def include_in_catch(self, loggername, include_children):
"""include a
:param bool include_children: also include children of the logger?
"""
self.fed_loggers.add(loggername)
if include_children:
self.fed_logger_children.add(loggername)
@classmethod
def get_wrapper(cls, stream_name, stream, *args, **kwargs):
if stream_name in cls.INSTANCES:
wrapper, creation_args = cls.INSTANCES[stream_name]
if creation_args != (args, kwargs):
warnings.warn(
"Existing instance of {} was created with args {}"
"and you are trying to fetch it with different args: {}"
"".format(cls, creation_args, (args, kwargs))
)
return wrapper, False
wrapper = cls(stream, *args, **kwargs)
cls.INSTANCES[stream_name] = wrapper, (args, kwargs)
return wrapper, True