如何在Python中将stdout和stderr重定向到记录器

Question

我有一个带有

RotatingFileHandler

的记录器。 我想将所有

Stdout

和

Stderr

重定向到记录器。 如何做到这一点？

Answer 1

没有足够的代表发表评论，但我想添加对我有用的版本，以防其他人遇到类似情况。

class LoggerWriter:
    def __init__(self, level):
        # self.level is really like using log.debug(message)
        # at least in my case
        self.level = level

    def write(self, message):
        # if statement reduces the amount of newlines that are
        # printed to the logger
        if message != '\n':
            self.level(message)

    def flush(self):
        # create a flush method so things can be flushed when
        # the system wants to. Not sure if simply 'printing'
        # sys.stderr is the correct way to do it, but it seemed
        # to work properly for me.
        self.level(sys.stderr)

这看起来像：

log = logging.getLogger('foobar')
sys.stdout = LoggerWriter(log.debug)
sys.stderr = LoggerWriter(log.warning)

Answer 2

Python 3 更新：

• 包括一个虚拟刷新函数，可以防止在预期函数中出现错误（Python 2 只需

  linebuf=''

  就可以了）。
• 请注意，如果从解释器会话记录的输出（和日志级别）与从文件运行的输出（和日志级别）显示不同。从文件运行会产生预期的行为（以及下面的输出）。
• 我们仍然消除了其他解决方案所没有的额外换行符。

class StreamToLogger(object):
    """
    Fake file-like stream object that redirects writes to a logger instance.
    """
    def __init__(self, logger, level):
       self.logger = logger
       self.level = level
       self.linebuf = ''

    def write(self, buf):
       for line in buf.rstrip().splitlines():
          self.logger.log(self.level, line.rstrip())

    def flush(self):
        pass

然后用类似的东西进行测试：

import StreamToLogger
import sys
import logging

logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s:%(levelname)s:%(name)s:%(message)s',
        filename='out.log',
        filemode='a'
        )
log = logging.getLogger('foobar')
sys.stdout = StreamToLogger(log,logging.INFO)
sys.stderr = StreamToLogger(log,logging.ERROR)
print('Test to standard out')
raise Exception('Test to standard error')

请参阅下面的旧 Python 2.x 答案和示例输出：

之前的所有答案似乎都存在在不需要的地方添加额外换行符的问题。最适合我的解决方案来自 http://www.electricmonk.nl/log/2011/08/14/redirect-stdout-and-stderr-to-a-logger-in-python/，他在其中演示如何将 stdout 和 stderr 发送到记录器：

import logging
import sys
 
class StreamToLogger(object):
   """
   Fake file-like stream object that redirects writes to a logger instance.
   """
   def __init__(self, logger, log_level=logging.INFO):
      self.logger = logger
      self.log_level = log_level
      self.linebuf = ''
 
   def write(self, buf):
      for line in buf.rstrip().splitlines():
         self.logger.log(self.log_level, line.rstrip())
 
logging.basicConfig(
   level=logging.DEBUG,
   format='%(asctime)s:%(levelname)s:%(name)s:%(message)s',
   filename="out.log",
   filemode='a'
)
 
stdout_logger = logging.getLogger('STDOUT')
sl = StreamToLogger(stdout_logger, logging.INFO)
sys.stdout = sl
 
stderr_logger = logging.getLogger('STDERR')
sl = StreamToLogger(stderr_logger, logging.ERROR)
sys.stderr = sl
 
print "Test to standard out"
raise Exception('Test to standard error')

输出如下：

2011-08-14 14:46:20,573:INFO:STDOUT:Test to standard out
2011-08-14 14:46:20,573:ERROR:STDERR:Traceback (most recent call last):
2011-08-14 14:46:20,574:ERROR:STDERR:  File "redirect.py", line 33, in 
2011-08-14 14:46:20,574:ERROR:STDERR:raise Exception('Test to standard error')
2011-08-14 14:46:20,574:ERROR:STDERR:Exception
2011-08-14 14:46:20,574:ERROR:STDERR::
2011-08-14 14:46:20,574:ERROR:STDERR:Test to standard error

请注意，

self.linebuf = ''

是处理刷新的地方，而不是实现刷新函数。

Answer 3

如果它是一个全 Python 系统（即没有 C 库直接写入 fds，正如 Ignacio Vazquez-Abrams 所询问的那样），那么您可能可以使用here:

建议的方法

class LoggerWriter:
    def __init__(self, logger, level):
        self.logger = logger
        self.level = level

    def write(self, message):
        if message != '\n':
            self.logger.log(self.level, message)

然后将

sys.stdout

和

sys.stderr

设置为

LoggerWriter

实例。

Answer 4

您可以使用redirect_stdout上下文管理器：

import logging
from contextlib import redirect_stdout

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
logging.write = lambda msg: logging.info(msg) if msg != '\n' else None

with redirect_stdout(logging):
    print('Test')

或者像这样

import logging
from contextlib import redirect_stdout


logger = logging.getLogger('Meow')
logger.setLevel(logging.INFO)
formatter = logging.Formatter(
    fmt='[{name}] {asctime} {levelname}: {message}',
    datefmt='%m/%d/%Y %H:%M:%S',
    style='{'
)
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
ch.setFormatter(formatter)
logger.addHandler(ch)

logger.write = lambda msg: logger.info(msg) if msg != '\n' else None

with redirect_stdout(logger):
    print('Test')

Answer 5

输出重定向正确完成！

问题

logger.log

和其他函数（

.info

/

.error

/等）将每个调用输出为单独的行，即隐式添加（格式化和）换行符。

另一方面，

sys.stderr.write

只是将其文字输入写入流，包括部分行。例如：输出“ZeroDivisionError：除以零”实际上是对

sys.stderr.write

:

的 4(!) 次单独调用

sys.stderr.write('ZeroDivisionError')
sys.stderr.write(': ')
sys.stderr.write('division by zero')
sys.stderr.write('\n')

最受支持的 4 种方法（1、2、3、4）因此会产生额外的换行符 - 只需将“1/0”放入您的程序中，您将得到以下结果：

2021-02-17 13:10:40,814 - ERROR - ZeroDivisionError
2021-02-17 13:10:40,814 - ERROR - : 
2021-02-17 13:10:40,814 - ERROR - division by zero

解决方案

将中间写入存储在缓冲区中。我使用列表而不是字符串作为缓冲区的原因是为了避免Shlemiel 画家的算法。 TLDR：它是 O(n) 而不是潜在的 O(n^2)

class LoggerWriter:
    def __init__(self, logfct):
        self.logfct = logfct
        self.buf = []

    def write(self, msg):
        if msg.endswith('\n'):
            self.buf.append(msg.removesuffix('\n'))
            self.logfct(''.join(self.buf))
            self.buf = []
        else:
            self.buf.append(msg)

    def flush(self):
        pass

# To access the original stdout/stderr, use sys.__stdout__/sys.__stderr__
sys.stdout = LoggerWriter(logger.info)
sys.stderr = LoggerWriter(logger.error)

2021-02-17 13:15:22,956 - ERROR - ZeroDivisionError: division by zero

对于 Python 3.9 以下的版本，您可以将

msg.removesuffix('\n')

替换为

msg.rstrip('\n')

或

msg[:-1]

。

Answer 6

作为 Cameron Gagnon 响应的演变，我将

LoggerWriter

类改进为：

class LoggerWriter(object):
    def __init__(self, writer):
        self._writer = writer
        self._msg = ''

    def write(self, message):
        self._msg = self._msg + message
        while '\n' in self._msg:
            pos = self._msg.find('\n')
            self._writer(self._msg[:pos])
            self._msg = self._msg[pos+1:]

    def flush(self):
        if self._msg != '':
            self._writer(self._msg)
            self._msg = ''

现在不受控制的异常看起来更好了：

2018-07-31 13:20:37,482 - ERROR - Traceback (most recent call last):
2018-07-31 13:20:37,483 - ERROR -   File "mf32.py", line 317, in <module>
2018-07-31 13:20:37,485 - ERROR -     main()
2018-07-31 13:20:37,486 - ERROR -   File "mf32.py", line 289, in main
2018-07-31 13:20:37,488 - ERROR -     int('')
2018-07-31 13:20:37,489 - ERROR - ValueError: invalid literal for int() with base 10: ''

Answer 7

快速但脆弱的单线

sys.stdout.write = logger.info

sys.stderr.write = logger.error

它的作用只是将记录器函数分配给 stdout/stderr

.write

调用，这意味着任何写入调用都将调用记录器函数。

这种方法的缺点是，对

.write

的调用和记录器函数通常都会添加换行符，因此您最终会在日志文件中出现额外的行，这可能会或可能不会成为问题，具体取决于您的用例。

另一个陷阱是，如果您的记录器写入 stderr 本身，我们会得到无限递归（堆栈溢出错误）。所以只输出到文件。

Answer 8

Vinay Sajip 的答案中添加了同花：

class LoggerWriter:
    def __init__(self, logger, level): 
        self.logger = logger
        self.level = level 

    def write(self, message):
        if message != '\n':
            self.logger.log(self.level, message)

    def flush(self): 
        pass

Answer 9

解决StreamHandler导致无限递归的问题

我的记录器导致无限递归，因为 Streamhandler 试图写入标准输出，它本身就是一个记录器 -> 导致无限递归。

解决方案

仅恢复 StreamHandler 的原始

sys.__stdout__

，以便您仍然可以看到终端中显示的日志。

class DefaultStreamHandler(logging.StreamHandler):
    def __init__(self, stream=sys.__stdout__):
        # Use the original sys.__stdout__ to write to stdout
        # for this handler, as sys.stdout will write out to logger.
        super().__init__(stream)


class LoggerWriter(io.IOBase):
    """Class to replace the stderr/stdout calls to a logger"""

    def __init__(self, logger_name: str, log_level: int):
        """:param logger_name: Name to give the logger (e.g. 'stderr')
        :param log_level: The log level, e.g. logging.DEBUG / logging.INFO that
                          the MESSAGES should be logged at.
        """
        self.std_logger = logging.getLogger(logger_name)
        # Get the "root" logger from by its name (i.e. from a config dict or at the bottom of this file)
        #  We will use this to create a copy of all its settings, except the name
        app_logger = logging.getLogger("myAppsLogger")
        [self.std_logger.addHandler(handler) for handler in app_logger.handlers]
        self.std_logger.setLevel(app_logger.level)  # the minimum lvl msgs will show at
        self.level = log_level  # the level msgs will be logged at
        self.buffer = []

    def write(self, msg: str):
        """Stdout/stderr logs one line at a time, rather than 1 message at a time.
        Use this function to aggregate multi-line messages into 1 log call."""
        msg = msg.decode() if issubclass(type(msg), bytes) else msg

        if not msg.endswith("\n"):
            return self.buffer.append(msg)

        self.buffer.append(msg.rstrip("\n"))
        message = "".join(self.buffer)
        self.std_logger.log(self.level, message)
        self.buffer = []


def replace_stderr_and_stdout_with_logger():
    """Replaces calls to sys.stderr -> logger.info & sys.stdout -> logger.error"""
    # To access the original stdout/stderr, use sys.__stdout__/sys.__stderr__
    sys.stdout = LoggerWriter("stdout", logging.INFO)
    sys.stderr = LoggerWriter("stderr", logging.ERROR)


if __name__ == __main__():
    # Load the logger & handlers
    logger = logging.getLogger("myAppsLogger")
    logger.setLevel(logging.DEBUG)
    # HANDLER = logging.StreamHandler()
    HANDLER = DefaultStreamHandler()  # <--- replace the normal streamhandler with this
    logger.addHandler(HANDLER)
    logFormatter = logging.Formatter("[%(asctime)s] - %(name)s - %(levelname)s - %(message)s")
    HANDLER.setFormatter(logFormatter)

    # Run this AFTER you load the logger
    replace_stderr_and_stdout_with_logger()

初始化记录器后最后调用

replace_stderr_and_stdout_with_logger()

（代码的最后一点）

Answer 10

如果您想将信息和错误消息记录到单独的流中（信息记录到 stdout，错误记录到 stderr），您可以使用以下技巧：

class ErrorStreamHandler(log.StreamHandler):
"""Print input log-message into stderr, print only error/warning messages"""
def __init__(self, stream=sys.stderr):
    log.Handler.__init__(self, log.WARNING)
    self.stream = stream

def emit(self, record):
    try:
        if record.levelno in (log.INFO, log.DEBUG, log.NOTSET):
            return
        msg = self.format(record)
        stream = self.stream
        # issue 35046: merged two stream.writes into one.
        stream.write(msg + self.terminator)
        self.flush()
    except RecursionError:  # See issue 36272
        raise
    except Exception:
        self.handleError(record)


class OutStreamHandler(log.StreamHandler):
"""Print input log-message into stdout, print only info/debug messages"""
def __init__(self, loglevel, stream=sys.stdout):
    log.Handler.__init__(self, loglevel)
    self.stream = stream

def emit(self, record):
    try:
        if record.levelno not in (log.INFO, log.DEBUG, log.NOTSET):
            return
        msg = self.format(record)
        stream = self.stream
        # issue 35046: merged two stream.writes into one.
        stream.write(msg + self.terminator)
        self.flush()
    except RecursionError:  # See issue 36272
        raise
    except Exception:
        self.handleError(record)

用途：

log.basicConfig(level=settings.get_loglevel(),
                    format="[%(asctime)s] %(levelname)s: %(message)s",
                    datefmt='%Y/%m/%d %H:%M:%S', handlers=[ErrorStreamHandler(), OutStreamHandler(settings.get_loglevel())])

Answer 11

检查这个解决方案，它对我有用。

在下面的 StreamToLogger 类中，您将记录器实例和日志级别传递给该类的每个实例。 stdout 和 stderr 的日志级别可以不同。

此方法提供了更大的灵活性，因为您可以在创建每个实例时动态决定日志级别。如果您想要不同级别的 stdout 和 stderr，这非常有用。并且 write 方法直接使用指定的日志级别记录消息。

class StreamToLogger:

    def __init__(self, logger, log_level=logging.INFO):
        self.logger = logger
        self.log_level = log_level

    def write(self, message):
        if message.rstrip() != "":
            self.logger.log(self.log_level, message.rstrip())

    def flush(self):
        pass


stdout_log_level_name = os.getenv('STDOUT_PYTHON_LOG_LEVEL', 'INFO').upper()
stdout_logging_level = getattr(logging, stdout_log_level_name, logging.INFO)
sys.stdout = StreamToLogger(custom_logger, stdout_logging_level)


stderr_log_level_name = os.getenv('STDERR_PYTHON_LOG_LEVEL', 'ERROR').upper()
stderr_logging_level = getattr(logging, stderr_log_level_name, logging.ERROR)
sys.stderr = StreamToLogger(custom_logger, stderr_logging_level)

custom_logger 是我的记录器，将其替换为您的记录器实例。

希望这有帮助， 快乐编码。