Skip to content

logger

Logging configuration and setup

MultiLineFormatter #

Bases: Formatter

Multi-line formatter with UUKID prepended.

Source code in src/sdss_explorer/util/logger.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class MultiLineFormatter(logging.Formatter):
    """Multi-line formatter with UUKID prepended."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def get_header_length(self, record):
        """Get the header length of a given record."""
        record_no_msg = logging.LogRecord(
            name=record.name,
            level=record.levelno,
            pathname=record.pathname,
            lineno=record.lineno,
            msg="",
            args=(),
            exc_info=None,
        )
        record_no_msg.kernel_id = record.kernel_id  # ensure this is copied properly
        return len(super().format(record_no_msg))

    def format(self, record):
        """Format a record with added indentation and custom property prepended."""
        # get header
        head, *trailing = super().format(record).splitlines(True)
        # first = record.getMessage().splitlines(True)[0]
        # head = head.replace(first, "")

        # Format the message and preserve multiline formatting
        indent = " " * (self.get_header_length(record))
        # indent = " "

        return head + "".join(indent + line for line in trailing)

format(record) #

Format a record with added indentation and custom property prepended.

Source code in src/sdss_explorer/util/logger.py 
31
32
33
34
35
36
37
38
39
40
41
42
def format(self, record):
    """Format a record with added indentation and custom property prepended."""
    # get header
    head, *trailing = super().format(record).splitlines(True)
    # first = record.getMessage().splitlines(True)[0]
    # head = head.replace(first, "")

    # Format the message and preserve multiline formatting
    indent = " " * (self.get_header_length(record))
    # indent = " "

    return head + "".join(indent + line for line in trailing)

get_header_length(record) #

Get the header length of a given record.

Source code in src/sdss_explorer/util/logger.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
def get_header_length(self, record):
    """Get the header length of a given record."""
    record_no_msg = logging.LogRecord(
        name=record.name,
        level=record.levelno,
        pathname=record.pathname,
        lineno=record.lineno,
        msg="",
        args=(),
        exc_info=None,
    )
    record_no_msg.kernel_id = record.kernel_id  # ensure this is copied properly
    return len(super().format(record_no_msg))

get_kernel_id() #

Fetches kernel ID per context

Source code in src/sdss_explorer/util/logger.py 
45
46
47
48
49
50
def get_kernel_id() -> str:
    """Fetches kernel ID per context"""
    try:
        return str(sl.get_kernel_id())
    except Exception:
        return "dummy"

setup_logging(log_path='./', log_file='explorerApp.log', console_log_level='DEBUG', file_log_level='INFO') #

Configures the logging system with a rotating file handler.

Parameters:

Name Type Description Default
log_path str

log path

 './'
log_file str

log filename

 'explorerApp.log'
console_log_level str

log level for console as full uppercase string

 'DEBUG'
file_log_level str

log level for file as full uppercase string

 'INFO'
Source code in src/sdss_explorer/util/logger.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def setup_logging(
    log_path: str = "./",
    log_file: str = "explorerApp.log",
    console_log_level: str = "DEBUG",
    file_log_level: str = "INFO",
):
    """
    Configures the logging system with a rotating file handler.

    Args:
        log_path: log path
        log_file: log filename
        console_log_level: log level for console as full uppercase string
        file_log_level: log level for file as full uppercase string
    """
    logging_config = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "standard": {
                "()":
                # logging.Formatter,
                MultiLineFormatter,  # use custom multi-line formatter
                "format":
                "%(asctime)s - %(name)s - %(levelname)s - %(kernel_id)s - %(message)s",  # standard format
            }
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "formatter": "standard",
                "level": console_log_level,
            },
            "file": {
                "class": "logging.FileHandler",
                "formatter": "standard",
                "level": file_log_level,
                "filename": pathjoin(log_path, log_file),
                "mode": "a",
            },
        },
        "loggers": {
            "dashboard": {
                "handlers": ["console", "file"],
                "level": file_log_level,
                "propagate": False,
            },
            "server": {
                "handlers": ["console", "file"],
                "level": file_log_level,
                "propagate": False,
            },
        },
    }

    # set record factory to set kernel id
    oldfactory = logging.getLogRecordFactory()

    def record_factory(*args, **kwargs):
        record = oldfactory(*args, **kwargs)
        if getattr(record, "kernel_id", None) is None:
            record.kernel_id = get_kernel_id()
        return record

    logging.setLogRecordFactory(record_factory)

    logging.config.dictConfig(logging_config)
    return