#6-2020-Feb-“Logging”

How to format your console output?

import logging
format = "%(asctime)s.%(msecs)03d %(levelname)-7s %(threadName)-13s %(name)42s.%(lineno)-6s'%(message)-s'"
DATE_FMT_SHORT = "%M:%S"
DATE_FMT_LONG = "%Y-%m-%d %H:%M:%S"
logging.basicConfig(level=logging.INFO, format=format, datefmt=DATE_FMT_SHORT)
awesome_logger = logging.getLogger('awesome_logger')
logger = logging.getLogger(__name__)
logger.info('my log message')

def thread_log():
    logger.info("thread_is_logging")

from multiprocessing.pool import ThreadPool
number_of_workers = 4
pool = ThreadPool(number_of_workers)
result = pool.apply_async(thread_log, callback=lambda *_: awesome_logger.info(f'complete!'))
result.wait(0.1)
logger.info(f'log demo complete')

outputs:

51:47.937 INFO    MainThread                                      __main__.8     'my log message'
51:47.987 INFO    Thread-1                                        __main__.10    'thread_is_logging'
51:47.987 INFO    Thread-7                                  awesome_logger.15    'complete!'
51:47.987 INFO    MainThread                                      __main__.17    'log demo complete'

Things to note

  1. The datefmt is short but msecs is included in format

  2. To use an explicit name, awesome_logger is not common. I use it to show how the name (LINE 3 above) affect the output

  3. -7s,-13s,-42s,-6s –> I use this to give space around/align messages. %(name)42s is the longest one (deep module hierarchy -> many characters)

How to format your console output when running pytest?

Use pytest.ini, read the great documentation!

[pytest]
log_format = %(asctime)s %(levelname)s %(message)s
log_date_format = %Y-%m-%d %H:%M:%S

Warning

Live logging is not as simple as you would think, see next point

Write directly to live output during pytest

This turns out to be harder than expected. Why? Pytest captures stdout and stderr to do checks and reporting. You need to temporary disable the capturing:

import logging
from contextlib import contextmanager

from utils.closable_queue import ClosableQueue
from utils.pool_async import apply_in_pool

logger = logging.getLogger(__name__)

@contextmanager
def pytest_live_logger():
    root = logging.getLogger()
    handler_1 = root.handlers[0]
    assert hasattr(handler_1, "capture_manager"), "handler did not have capture manager"
    # This is ensuring stdout is not captured
    with handler_1.capture_manager.global_and_fixture_disabled(): 
        yield handler_1.stream

def log_to_live_output(out: ClosableQueue, name: str):
    with pytest_live_logger() as out_stream:
        for m in out:
            out_stream.write(f"{name}: {m}")

# Example test:
def test_live_logging():
    queue = ClosableQueue(maxsize=1)
    logger.info(f"live logging")
    task = apply_in_pool(log_to_live_output, queue, "live_logger.py") # just a wrapper
    for i in range(1, 4):
        queue.put(f"message: {i}\n")
        logger.info(f"sending message:  {i}")
    queue.close()
    task.wait(timeout=0.01)
    logger.info(f"test complete")

Test live-log, result 1

56:08.463 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'live logging'
56:08.463 INFO     MainThread                                                     utils.pool_async.apply_in_pool         'log_to_live_output((<utils.closable_queue.ClosableQueue object at 0x11adf7fd0>, 'live_logger.py'), '
56:08.463 INFO     MainThread                                                     utils.pool_async.get_pool              'number of workers: 20'
56:08.475 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'sending message:  1'
live_logger.py: message: 1
56:08.476 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'sending message:  2'
live_logger.py: message: 3
56:08.476 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'sending message:  3'
56:08.476 INFO     Thread-23                                                 utils.thread_complete.complete_callback     'log_to_live_output((<utils.closable_queue.ClosableQueue object at 0x11adf7fd0>, 'live_logger.py'),  completed'
56:08.477 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'test complete'
PASSEDlive_logger.py: message: 2

Test live-log, result 2

57:55.311 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'live logging'
57:55.311 INFO     MainThread                                                     utils.pool_async.apply_in_pool         'log_to_live_output((<utils.closable_queue.ClosableQueue object at 0x10e9b2fd0>, 'live_logger.py'), '
57:55.312 INFO     MainThread                                                     utils.pool_async.get_pool              'number of workers: 20'
57:55.328 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'sending message:  1'
live_logger.py: message: 2
57:55.329 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'sending message:  3'
live_logger.py: message: 3
57:55.329 INFO     Thread-23                                                 utils.thread_complete.complete_callback     'log_to_live_output((<utils.closable_queue.ClosableQueue object at 0x10e9b2fd0>, 'live_logger.py'),  completed'
57:55.330 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'test complete'
PASSEDlive_logger.py: message: 1
57:55.329 INFO     MainThread                           tests_flow.test_scneario_localhost_fixture.test_live_logging     'sending message:  2'

Note

The order of the log messages is not consistent! message: 2 is not logged until after the test for result 1 and the result 2 is even more weird! I am not sure why. Please let me know if you do!

Discussion: How to get a single (ordered) source of output when logging in python?

Have you ever had problems with unordered log messages? (e.g., see previous point) Ideas:

  1. Implement a log server that receives the log records and outputs directly to stdout after a short delay. (Needs to parse ts of log messages). Can be done by extending logging.Handler

  2. Deactivate all the capturing from pytest in you pytest.ini (easy fix)

addopts = -s --show-capture=no -p no:logging -p no:warnings

How to you write an error message with traceback from outside of the exception handler?

import logging
import traceback

def log_error(logger, error, level=logging.ERROR):
    logger.log(level, str(error))
    traceback__ = getattr(error, "__traceback__", "")
    if traceback__:
        logger.log(level, "".join(traceback.format_tb(traceback__)))

Discussion: How to write log messages?