added logging sub-module

added logging sub-module python 2 compatibility fixed python 2 fix added test for custom tqdm class python 2 absolute imports (due to otherwise conflicting `logging` module) isort more tests relating to _get_first_found_console_formatter isort minor simplification test handleError test logging formatter being used minor rename to _get_first_found_console_logging_formatter test that certain exceptions are not swallowed avoid using mock.assert_called (py 3.5) moved to tqdm.contrib.logging added "Redirecting console logging to tqdm" readme removed no longer necessary absolute_import declaration minor: updated package of example in docstring
KOLANICH-libs · Apr 5, 2021 · e86b9b1 · e86b9b1
1 parent 9fb0f23
commit e86b9b1
Show file tree

Hide file tree

Showing 4 changed files with 429 additions and 0 deletions.
diff --git a/README.rst b/README.rst
@@ -1321,6 +1321,59 @@ A reusable canonical example is given below:
     # After the `with`, printing is restored
     print("Done!")
 
+Redirecting console logging to tqdm
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Similar to redirecting ``sys.stdout`` directly as detailed in the previous section,
+you may want to redirect logging that would otherwise go to the
+console (``sys.stdout`` or ``sys.stderr``) to ``tqdm``.
+
+Note: if you are also replace ``sys.stdout`` and ``sys.stderr`` at the same time,
+then the logging should be redirected first. Otherwise it won't be able to detect
+the console logging handler.
+
+For that you may use ``redirect_logging_to_tqdm`` or ``tqdm_with_logging_redirect``
+from ``tqdm.contrib.logging``. Both methods accept the following optional parameters:
+
+- ``loggers``: A list of loggers to update. Defaults to ``logging.root``.
+- ``tqdm``: A ``tqdm`` class. Defaults to ``tqdm.tqdm``.
+
+An example redirecting the console logging to tqdm:
+
+.. code:: python
+
+    import logging
+    from tqdm.contrib.logging import redirect_logging_to_tqdm
+
+    LOGGER = logging.getLogger(__name__)
+
+    if __name__ == '__main__':
+        logging.basicConfig(level='INFO')
+        with redirect_logging_to_tqdm():
+            # logging to the console is now redirected to tqdm
+            LOGGER.info('some message')
+        # logging is now restored
+
+An similar example, wrapping tqdm while redirecting console logging:
+
+.. code:: python
+
+    import logging
+    from tqdm.contrib.logging import tqdm_with_logging_redirect
+
+    LOGGER = logging.getLogger(__name__)
+
+    if __name__ == '__main__':
+        logging.basicConfig(level='INFO')
+
+        file_list = ['file1', 'file2']
+        with tqdm_with_logging_redirect(total=len(file_list)) as pbar:
+            # logging to the console is now redirected to tqdm
+            for filename in file_list:
+                LOGGER.info('processing file: %s', filename)
+                pbar.update(1)
+        # logging is now restored
+
 Monitoring thread, intervals and miniters
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

diff --git a/tests/contrib/__init__.py b/tests/contrib/__init__.py
diff --git a/tests/contrib/tests_logging.py b/tests/contrib/tests_logging.py
@@ -0,0 +1,182 @@
+# pylint: disable=missing-module-docstring, missing-class-docstring
+# pylint: disable=missing-function-docstring, no-self-use
+
+from __future__ import absolute_import
+
+import logging
+import logging.handlers
+import sys
+from io import StringIO
+
+import pytest
+
+from tqdm import tqdm
+from tqdm.contrib.logging import _get_first_found_console_logging_formatter
+from tqdm.contrib.logging import _TqdmLoggingHandler as TqdmLoggingHandler
+from tqdm.contrib.logging import redirect_logging_to_tqdm, tqdm_with_logging_redirect
+
+from ..tests_tqdm import importorskip
+
+LOGGER = logging.getLogger(__name__)
+
+TEST_LOGGING_FORMATTER = logging.Formatter()
+
+
+class CustomTqdm(tqdm):
+    messages = []
+
+    @classmethod
+    def write(cls, s, **__):  # pylint: disable=arguments-differ
+        CustomTqdm.messages.append(s)
+
+
+class ErrorRaisingTqdm(tqdm):
+    exception_class = RuntimeError
+
+    @classmethod
+    def write(cls, s, **__):  # pylint: disable=arguments-differ
+        raise ErrorRaisingTqdm.exception_class('fail fast')
+
+
+class TestTqdmLoggingHandler:
+    def test_should_call_tqdm_write(self):
+        CustomTqdm.messages = []
+        logger = logging.Logger('test')
+        logger.handlers = [TqdmLoggingHandler(CustomTqdm)]
+        logger.info('test')
+        assert CustomTqdm.messages == ['test']
+
+    def test_should_call_handle_error_if_exception_was_thrown(self):
+        patch = importorskip('unittest.mock').patch
+        logger = logging.Logger('test')
+        ErrorRaisingTqdm.exception_class = RuntimeError
+        handler = TqdmLoggingHandler(ErrorRaisingTqdm)
+        logger.handlers = [handler]
+        with patch.object(handler, 'handleError') as mock:
+            logger.info('test')
+            assert mock.called
+
+    @pytest.mark.parametrize('exception_class', [
+        KeyboardInterrupt,
+        SystemExit
+    ])
+    def test_should_not_swallow_certain_exceptions(self, exception_class):
+        logger = logging.Logger('test')
+        ErrorRaisingTqdm.exception_class = exception_class
+        handler = TqdmLoggingHandler(ErrorRaisingTqdm)
+        logger.handlers = [handler]
+        with pytest.raises(exception_class):
+            logger.info('test')
+
+
+class TestGetFirstFoundConsoleLoggingFormatter:
+    def test_should_return_none_for_no_handlers(self):
+        assert _get_first_found_console_logging_formatter([]) is None
+
+    def test_should_return_none_without_stream_handler(self):
+        handler = logging.handlers.MemoryHandler(capacity=1)
+        handler.formatter = TEST_LOGGING_FORMATTER
+        assert _get_first_found_console_logging_formatter([handler]) is None
+
+    def test_should_return_none_for_stream_handler_not_stdout_or_stderr(self):
+        handler = logging.StreamHandler(StringIO())
+        handler.formatter = TEST_LOGGING_FORMATTER
+        assert _get_first_found_console_logging_formatter([handler]) is None
+
+    def test_should_return_stream_handler_formatter_if_stream_is_stdout(self):
+        handler = logging.StreamHandler(sys.stdout)
+        handler.formatter = TEST_LOGGING_FORMATTER
+        assert _get_first_found_console_logging_formatter(
+            [handler]
+        ) == TEST_LOGGING_FORMATTER
+
+    def test_should_return_stream_handler_formatter_if_stream_is_stderr(self):
+        handler = logging.StreamHandler(sys.stderr)
+        handler.formatter = TEST_LOGGING_FORMATTER
+        assert _get_first_found_console_logging_formatter(
+            [handler]
+        ) == TEST_LOGGING_FORMATTER
+
+
+class TestRedirectLoggingToTqdm:
+    def test_should_add_and_remove_tqdm_handler(self):
+        logger = logging.Logger('test')
+        with redirect_logging_to_tqdm(loggers=[logger]):
+            assert len(logger.handlers) == 1
+            assert isinstance(logger.handlers[0], TqdmLoggingHandler)
+        assert not logger.handlers
+
+    def test_should_remove_and_restore_console_handlers(self):
+        logger = logging.Logger('test')
+        stderr_console_handler = logging.StreamHandler(sys.stderr)
+        stdout_console_handler = logging.StreamHandler(sys.stderr)
+        logger.handlers = [stderr_console_handler, stdout_console_handler]
+        with redirect_logging_to_tqdm(loggers=[logger]):
+            assert len(logger.handlers) == 1
+            assert isinstance(logger.handlers[0], TqdmLoggingHandler)
+        assert logger.handlers == [stderr_console_handler, stdout_console_handler]
+
+    def test_should_inherit_console_logger_formatter(self):
+        logger = logging.Logger('test')
+        formatter = logging.Formatter('custom: %(message)s')
+        console_handler = logging.StreamHandler(sys.stderr)
+        console_handler.setFormatter(formatter)
+        logger.handlers = [console_handler]
+        with redirect_logging_to_tqdm(loggers=[logger]):
+            assert logger.handlers[0].formatter == formatter
+
+    def test_should_not_remove_stream_handlers_not_fot_stdout_or_stderr(self):
+        logger = logging.Logger('test')
+        stream_handler = logging.StreamHandler(StringIO())
+        logger.addHandler(stream_handler)
+        with redirect_logging_to_tqdm(loggers=[logger]):
+            assert len(logger.handlers) == 2
+            assert logger.handlers[0] == stream_handler
+            assert isinstance(logger.handlers[1], TqdmLoggingHandler)
+        assert logger.handlers == [stream_handler]
+
+
+class TestTqdmWithLoggingRedirect:
+    def test_should_add_and_remove_handler_from_root_logger_by_default(self):
+        original_handlers = list(logging.root.handlers)
+        with tqdm_with_logging_redirect(total=1) as pbar:
+            assert isinstance(logging.root.handlers[-1], TqdmLoggingHandler)
+            LOGGER.info('test')
+            pbar.update(1)
+        assert logging.root.handlers == original_handlers
+
+    def test_should_add_and_remove_handler_from_custom_logger(self):
+        logger = logging.Logger('test')
+        with tqdm_with_logging_redirect(total=1, loggers=[logger]) as pbar:
+            assert len(logger.handlers) == 1
+            assert isinstance(logger.handlers[0], TqdmLoggingHandler)
+            logger.info('test')
+            pbar.update(1)
+        assert not logger.handlers
+
+    def test_should_not_fail_with_logger_without_console_handler(self):
+        logger = logging.Logger('test')
+        logger.handlers = []
+        with tqdm_with_logging_redirect(total=1, loggers=[logger]):
+            logger.info('test')
+        assert not logger.handlers
+
+    def test_should_format_message(self):
+        logger = logging.Logger('test')
+        console_handler = logging.StreamHandler(sys.stdout)
+        console_handler.setFormatter(logging.Formatter(
+            r'prefix:%(message)s'
+        ))
+        logger.handlers = [console_handler]
+        CustomTqdm.messages = []
+        with tqdm_with_logging_redirect(loggers=[logger], tqdm=CustomTqdm):
+            logger.info('test')
+        assert CustomTqdm.messages == ['prefix:test']
+
+    def test_use_root_logger_by_default_and_write_to_custom_tqdm(self):
+        logger = logging.root
+        CustomTqdm.messages = []
+        with tqdm_with_logging_redirect(total=1, tqdm=CustomTqdm) as pbar:
+            assert isinstance(pbar, CustomTqdm)
+            logger.info('test')
+            assert CustomTqdm.messages == ['test']