Python Loguru Hacking

loguru.md

Loguru is my favorite way to perform logging in python. It's generally so simple to use but a few things are missing from the documentation.

Logging over the network

This is based on the loguru example, but also includes the hostname of the sender. To add extra fields the pefered method is to add to record["extra"] dictionary (which is normally empty). The client code sends the log messages over the network and to stderr.

class LoggingSocketHandler:
    def __init__(self, host, port, timeout=600):
        """
        :param host: hostname
        :param port: port number
        :param timeout: If it can't connect after this many seconds throw a TimeoutError. If set to None, the
        timeout is infinite
        """
        self.host = host
        self.port = port
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.timeout = timeout
        self._origin = os.uname()[1]  # this is your local hostname
        self._connect()
       
    def _connect(self):
        start_time = time.perf_counter()
        absolute_start_time = start_time
        addr = (self.host, self.port)
        while True:
            try:
                self.sock = socket.create_connection(addr)
                break
            except OSError as ex:
                time.sleep(.5)
                if time.perf_counter() - start_time >= 5:
                    logger.warning('Waiting for logger server connection')
                    start_time = time.perf_counter()
                if self.timeout:
                    if time.perf_counter() - absolute_start_time >= self.timeout:
                        logger.error('Waiting for logger server connection')
                        raise TimeoutError(f'Can not connect to {self.host}:{self.port}')

    def write(self, message):
        record = message.record
        record['extra'].update({'host':self._origin}) # add the host hostname field
        data = pickle.dumps(record)
        slen = struct.pack(">L", len(data))
        self.sock.send(slen + data)

def _setup():
    host = 'loggerhost.org'
    logger.configure(handlers=[{"sink": LoggingSocketHandler(host, 9999)}])
    logger.add(sys.stderr) # optionally add the stderr printing

_setup()

# just to demo, remove these
logger.debug('Network client logger set up')
logger.info('demo info msg')
logger.warning('demo warning msg')
logger.error('demo error msg')

Here is my basic server code.

import socketserver
import pickle
import struct
import threading
from loguru import logger

class LoggingStreamHandler(socketserver.StreamRequestHandler):

  def handle(self):
        while True:
            chunk = self.connection.recv(4)
            if len(chunk) < 4:
                break
            slen = struct.unpack('>L', chunk)[0]
            chunk = self.connection.recv(slen)
            while len(chunk) < slen:
                chunk = chunk + self.connection.recv(slen - len(chunk))
            precord = pickle.loads(chunk)
            level, message = precord["level"].name, precord["message"]
            logger.patch(lambda record: record.update(precord)).log(level, message)

class LoggingServer:
    def __init__(self,host='localhost',port=9999):
        self.host=host
        self.port = port
        self.server = None
        self._server_thread=threading.Thread(target=self._start)

    def start(self):
        self._server_thread.start()
        return self

    def shutdown(self):
        self.server.shutdown()
        self._server_thread.join()

    def _start(self):
        logger.debug('Setting up server logger')
        self.server = socketserver.TCPServer((self.host,self.port), LoggingStreamHandler)
        self.server.serve_forever()
        logger.debug('Server logger done')

def formatter(record):
    fmt = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green>| <lvl>{level: <8}</lvl>| {name}:{function}:{line} {extra[host]} -  <lvl>{message}</lvl>"
    fmt_local = "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green>| <lvl>{level: <8}</lvl>| {name}:{function}:{line} local -  <lvl>{message}</lvl>"
    if 'host' in record["extra"]:
        return fmt
    else:
        return fmt_local


if __name__=="__main__":

    import signal
    import sys

    lg = LoggingServer().start()
    logger.remove()
    logger.add(sys.stderr,level="DEBUG",format=formatter)
    logger.add('tmp.log',format=formatter)  # also save to a file
    logger.info('Press Ctrl+C to exit after closing all clients.')
    def signal_handler(sig, frame):
        print('You pressed Ctrl+C!')
        lg.shutdown()
        sys.exit(0)
        
    signal.signal(signal.SIGINT, signal_handler)
    signal.pause() # wait for a signal

Note, how we need a custom formatter to show the extra record. This example runs until ^C is pressed.