Last active
October 28, 2025 10:58
-
-
Save mahdilamb/2e114317a9579edd5dbdc389d324973f to your computer and use it in GitHub Desktop.
Monkey patch of logger and print for streaming output of python methods elsewhere
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 typing import Optional, Callable, Any, List | |
| import threading | |
| _threads: Optional[List[threading.Thread]] = None | |
| def patch_outputs(command: Callable[[Any, ...], None] = print, multithreaded: bool = True): | |
| """ | |
| Method that patches the logging and print methods in python | |
| :param multithreaded: whether to run the command in a separate thread | |
| :param command: the command to use in addition to the default | |
| :return: Nothing | |
| """ | |
| global _threads | |
| if _threads is not None: | |
| raise RuntimeError("Trying to patch outputs multiple times") | |
| import logging | |
| import builtins | |
| _threads = [] | |
| _OriginalLogger = logging.Logger | |
| _original_print = builtins.print | |
| def output_response(level: int, *args): | |
| def wrapped_command(*_args): | |
| """ | |
| Wrap the command so that the original logger and print work and don't create infinite recurstion | |
| :param _args: the arguments passed to the command | |
| :return: None | |
| """ | |
| builtins.print = _original_print | |
| logging.getLogger = logging.Logger = _OriginalLogger | |
| command(*_args) | |
| logging.getLogger = logging.Logger = _Logger | |
| builtins.print = _print | |
| if multithreaded: | |
| t = threading.Thread(target=wrapped_command, args=(level, " ".join((str(a) for a in args)))) | |
| _threads.append(t) | |
| t.start() | |
| else: | |
| wrapped_command(level, " ".join((str(a) for a in args))) | |
| class _Logger(_OriginalLogger): | |
| def _log(self, *args, **kwargs) -> None: | |
| super()._log(*args, **kwargs) | |
| output_response(*args[:2]) | |
| def _print(*args, **kwargs): | |
| _original_print(*args, **kwargs) | |
| output_response(logging.NOTSET, *args) | |
| # modify constructor and factory method | |
| logging.getLogger = logging.Logger = _Logger | |
| # modify built-in print method | |
| builtins.print = _print | |
| if __name__ == "__main__": | |
| patch_outputs(print) | |
| import sys | |
| # remove self from arguments and run file | |
| sys.argv = sys.argv[1:] | |
| exec(open(sys.argv[0]).read()) | |
| # make sure all output have been sent | |
| if _threads: | |
| [t.join() for t in _threads if t] |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment