Python Logger

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# console handler
stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)


# file handler 
logging_folder = Path(yourfolder)
logging_fp = os.path.join(logging_folder, "your_logging_fp")
# unlink previous one 
Path(logging_fp).unlink(missing_ok=True)
file_handler = logging.FileHandler(logging_fp)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.removeHandler(file_handler)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 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
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 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

 import json
import logging
from logging.handlers import RotatingFileHandler
from logging import FileHandler
from tqdm.auto import tqdm
import multiprocessing as mp
from pythonjsonlogger import jsonlogger
import atexit
import os 
from utils import convert_partial_to_json

    
def setupLogger(loggername, loggingfile = None, logginglevel = logging.INFO, loggingqueue = None, file_handler_class=FileHandler, if_json=False):
    """
    Set up a logger with the specified name and optional logging file.

    Parameters:
    loggername (str): The name of the logger.
    loggingfile (str): The path to the logging file (optional).

    Returns:
    logging.Logger: The configured logger.
    """
    if loggername not in logging.Logger.manager.loggerDict:
        # Create a logger in the main process
        logger = logging.getLogger(loggername)
        logger.setLevel(logginglevel)
        streamhandler = logging.StreamHandler()
        if loggingfile is not None:
            if if_json:
                loggingfile = loggingfile.replace('.log', '.json')
                loggingfile = loggingfile + ".partial"
                # register itexit 
                atexit.register(convert_partial_to_json, loggingfile)
                
            if file_handler_class == FileHandler:
                filehandler = FileHandler(loggingfile, delay=True)
            elif file_handler_class == RotatingFileHandler:
                filehandler = RotatingFileHandler(loggingfile, maxBytes=1000000, backupCount=3, delay=True)
            else:
                raise ValueError(f"Unknown file_handler_class {file_handler_class}")
            processname = mp.current_process().name
            if processname == 'MainProcess':
                # clear the log file
                with open(loggingfile, 'w'):
                    pass
        formatter = logging.Formatter('%(asctime)s [%(processName)s] [%(levelname)s] %(message)s')
        streamhandler.setFormatter(formatter)
        if if_json:
            jsonformatter = jsonlogger.JsonFormatter('%(asctime)s [%(processName)s] [%(levelname)s] %(message)s')
            filehandler.setFormatter(jsonformatter)
        else:
            filehandler.setFormatter(formatter)
        logger.addHandler(streamhandler)
        if loggingfile is not None:
            logger.addHandler(filehandler)
        if loggingqueue is not None:
            queuehandler = logging.handlers.QueueHandler(loggingqueue)
            logger.addHandler(queuehandler)
        return logger
    else:
        return logging.getLogger(loggername)


class logging_tqdm(tqdm):
    def __init__(
            self,
            *args,
            loggername="logging_tqdm_logger",
            loggingfile = None,
            logginglevel = logging.INFO,
            loggingqueue = None,
            mininterval: float = 1,
            bar_format: str = '{desc}{percentage:3.0f}%{r_bar}',
            desc: str = 'progress: ',
            **kwargs):
        self._loggername = loggername
        self._loggingfile = loggingfile
        self._logginglevel = logginglevel
        self._loggingqueue = loggingqueue
        self._logger = setupLogger(loggername, loggingfile, logginglevel, loggingqueue)
        
        super().__init__(
            *args,
            mininterval=mininterval,
            bar_format=bar_format,
            desc=desc,
            **kwargs
        )

    @property
    def logger(self):
        if self._logger is not None:
            return self._logger
        return setupLogger(self._loggername, self._loggingfile, self._logginglevel, self._loggingqueue)

    def display(self, msg=None, pos=None):
        if not self.n:
            # skip progress bar before having processed anything
            return
        if not msg:
            msg = self.__str__()
        self.logger.info(f'%s', msg)