[Python] How to set up multi-thread logging to different loggers using QueueListener and...

I have a Python application that runs multiple tasks in parallel using ThreadPoolExecutor. I want all threads to log safely without race conditions. I also need two loggers:

1. 
   Console logger (prints messages to stdout)
   
   
2. 
   File logger (log structured JSON messages to output.log)
   

I understand that logging directly from threads to file or console can cause issues when multiple threads write at the same time. I read that using a QueueHandler in workers and a QueueListener is recommended. My setup looks roughly like this:

• setup_logging() creates:
  • a multiprocessing queue
  • console and JSON loggers
  • console and JSON handlers
  • QueueListener with both handlers
• worker(task) logs both JSON and console messages using the respective loggers.
• Tasks are submitted via ThreadPoolExecutor.

import logging
import logging.handlers
import multiprocessing
import concurrent.futures
import json

# ------------------------------------------------------------
# Setup logging
# ------------------------------------------------------------
def setup_logging():
log_queue = multiprocessing.Manager().Queue(-1)

# Console logger
console_logger = logging.getLogger("console_logger")
console_logger.setLevel(logging.DEBUG)
console_logger.addHandler(logging.handlers.QueueHandler(log_queue))

# JSON logger
json_logger = logging.getLogger("json_logger")
json_logger.setLevel(logging.INFO)
json_logger.addHandler(logging.handlers.QueueHandler(log_queue))

# Console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter("%(message)s"))

# JSON file handler
json_handler = logging.FileHandler("path/to/json.log")
json_handler.setFormatter(logging.Formatter("%(message)s"))

# Queue listener
listener = logging.handlers.QueueListener(
log_queue,
console_handler,
json_handler,
)

return listener, console_logger, json_logger

# ------------------------------------------------------------
# Worker function
# ------------------------------------------------------------
def worker(task, console_logger, json_logger):
try:
# perform some task
result = perform_task(task)

# log JSON result to JSON logger
json_logger.info(json.dumps(result))

# log human-readable message to console logger
console_logger.info("Task completed for: %s", task)

except Exception as e:
console_logger.error("Error processing task %s: %s", task, e)

# ------------------------------------------------------------
# Main program
# ------------------------------------------------------------
if __name__ == "__main__":
listener, console_logger, json_logger = setup_logging()
listener.start()

tasks = get_task_list() # abstracted task list

# Threaded execution
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [
executor.submit(worker, task, console_logger, json_logger)
for task in tasks
]

# wait for all threads to finish
for f in futures:
f.result()

listener.stop()


I want to know:

1. Is this a correct way to set up QueueHandler + QueueListener with multiple loggers?
2. What is the recommended pattern to safely separate JSON and console outputs in a multithreaded Python program?

Any guidance or examples would be appreciated.

Continue reading...