Skip to content

Logging

StreamToLogger 

StreamToLogger(logger, log_level=DEBUG)

Bases: TextIOBase

Fake file-like stream object that redirects writes to a logger instance.

StreamToLogger is made to extend TextIOBase to prevent error like below when running pytest with docker-compose as part of dockerised test in developing pyderivationagent package: AttributeError: 'StreamToLogger' object has no attribute 'isatty'

To reproduce the error, one may checkout to this commit and run pytest -s --docker-compose=./docker-compose.test.yml in the folder

This error was due to this line in pytest checking if sys.stdout.isatty() is True/False

Another fix is to provide "def isatty(self) -> bool:"" but extending TextIOBase seems to be a "safer"/"cleaner" fix, according to this comment

Source code in JPS_BASE_LIB/python_wrapper/twa/agentlogging/logging.py 
45
46
47
48
def __init__(self, logger, log_level=logging.DEBUG):
    self.logger = logger
    self.log_level = log_level
    self.linebuf = ''

get_logger 

get_logger(logger_name: str)

Get the dev or prod logger (avoids having to import 'logging' in calling code).

Parameters:

Name Type Description Default
logger_name str

name of the logger to be used, available options include 'dev' and 'prod'

 required

Returns:

Type Description

Logger to use for logging statements.
Source code in JPS_BASE_LIB/python_wrapper/twa/agentlogging/logging.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def get_logger(logger_name: str):
    """
    Get the dev or prod logger (avoids having to import 'logging' in calling code).

    Args:
        logger_name: name of the logger to be used, available options include 'dev' and 'prod'

    Returns:
        Logger to use for logging statements.
    """
    valid_logger_names = ['dev','prod']

    if logger_name in valid_logger_names:
        return logging.getLogger(logger_name)
    else:
        raise ValueError("Invalid logger name: allowed values are "+",".join(valid_logger_names))

shutdown 

shutdown()

Shutdown the logging system, should be called before application exit after all logging calls.

Source code in JPS_BASE_LIB/python_wrapper/twa/agentlogging/logging.py 
102
103
104
105
106
107
def shutdown():
    """
    Shutdown the logging system, should be called
    before application exit after all logging calls.
    """
    logging.shutdown()

clear_loggers 

clear_loggers()

Remove handlers from all loggers. Method adopted from this comment.

Source code in JPS_BASE_LIB/python_wrapper/twa/agentlogging/logging.py 
110
111
112
113
114
115
116
117
118
119
120
def clear_loggers():
    """
    Remove handlers from all loggers. Method adopted from
    [this comment](https://github.com/pytest-dev/pytest/issues/5502#issuecomment-647157873).
    """
    import logging
    loggers = [logging.getLogger()] + list(logging.Logger.manager.loggerDict.values())
    for logger in loggers:
        handlers = getattr(logger, 'handlers', [])
        for handler in handlers:
            logger.removeHandler(handler)