TL;DR dictConfig does not work with custom QueueHandler implementations in Python 3.12
Following the logging cookbook, I implemented a custom QueueHandler which uses ZMQ to send log records to a listener running in another process.
Here is the listener code which reads messages from the socket and passes them to the handler:
# zmq-logging.server.py
DEFAULT_ADDR = 'tcp://localhost:13231'
_context = zmq.Context()
atexit.register(_context.destroy, 0)
class ZMQQueueListener(logging.handlers.QueueListener):
def __init__(self, address, *handlers):
self.address = address
socket = _context.socket(zmq.PULL)
socket.bind(address)
super().__init__(socket, *handlers)
def dequeue(self, block: bool) -> logging.LogRecord:
data = self.queue.recv_json()
if data is None:
return None
return logging.makeLogRecord(data)
def enqueue_sentinel(self) -> None:
socket = _context.socket(zmq.PUSH)
socket.connect(self.address)
socket.send_json(None)
def stop(self) -> None:
super().stop()
self.queue.close(0)
if __name__ == '__main__':
listener = ZMQQueueListener(DEFAULT_ADDR, logging.StreamHandler())
listener.start()
print('Press Ctrl-C to stop.')
try:
while True: time.sleep(0.1)
finally:
listener.stop()
Here is the handler code which enqueues records in the zmq queue.
# zmq-logging-handler.py
_context = zmq.Context()
class ZMQQueueHandler(logging.handlers.QueueHandler):
def __init__(self, address, ctx=None):
self.ctx = ctx or _context
socket = self.ctx.socket(zmq.PUSH)
socket.connect(address)
super().__init__(socket)
def enqueue(self, record: logging.LogRecord) -> None:
self.queue.send_json(record.__dict__)
def close(self) -> None:
return self.queue.close()
And a dictionary config I used to configure the logger
config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'full': {
'format': "%(asctime)s %(levelname)-8s %(name)s %(message)s",
'datefmt': "%Y-%m-%d %H:%M:%S"
}
},
'handlers': {
'zmq-logging': {
'class': 'zmq-logging-handler.ZMQQueueHandler',
'formatter': 'full',
'level': 'DEBUG',
'address': DEFAULT_ADDR
}
},
'loggers': {
'': {
'level': 'DEBUG',
'propagate': False,
'handlers': ['zmq-logging']
}
}
}
However, it no longer works in Python 3.12. The configuration for the QueueHandlers changed as described in the logging.config. The handler config requires additional queue, listener and handlers parameters which I do not know how to use with my class model. Should I abandon extending QueueHandler or is there a way to fix the config dict? I don't understand how new listener property of the QueueHandler can be used if the listener and the handler need to be instantiated in separate processes.