Created
December 19, 2019 12:14
-
-
Save brake/768771ba12a9abab81763a7efd33250e to your computer and use it in GitHub Desktop.
IO wrapper for intercepting of stdout and stderr into log file managed by logging framework
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from contextlib import redirect_stderr, redirect_stdout | |
from logging import getLogger | |
from logging_io import LoggingErrorOutput, LoggingInfoOutput | |
logger = getLogger() | |
error_output = LoggingErrorOutput(logger) | |
normal_output = LoggingInfoOutput(logger) | |
with redirect_stdin(normal_output), redirect_stdout(error_output): | |
# run code writing to stdin or stdout | |
# all thah output will be intercepted by logger | |
# Enjoy! | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from abc import ABCMeta, abstractmethod | |
from io import TextIOBase | |
from logging import Logger | |
class LoggingOutput(TextIOBase, metaclass=ABCMeta): | |
def __init__(self, logger: Logger): | |
self.logger: Logger = logger | |
@abstractmethod | |
def write_to_log(self, message, *args, **kwargs) -> None: | |
... | |
def write(self, message: str, *args, **kwargs) -> None: | |
msg_cleared = message.strip() | |
if msg_cleared: | |
self.write_to_log(msg_cleared, *args, **kwargs) | |
def close(self) -> None: | |
pass | |
class LoggingErrorOutput(LoggingOutput): | |
def __init__(self, logger: Logger): | |
super().__init__(logger) | |
def write_to_log(self, message: str, *args, **kwargs) -> None: | |
self.logger.error(message, *args, **kwargs) | |
class LoggingInfoOutput(LoggingOutput): | |
def __init__(self, logger: Logger): | |
super().__init__(logger) | |
def write_to_log(self, message: str, *args, **kwargs) -> None: | |
self.logger.info(message, *args, **kwargs) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment