py-async-log 0.0.6

A high-performance multiprocess-safe asynchronous logging tool.一个支持多进程多线程的日志封装库

py-async-log

py-async-log is a high-performance, thread-safe, and multiprocess-safe asynchronous logging utility. It is specifically designed to solve log contention and blocking issues within Python's multithreading and multiprocessing environments.

Note: This is the initial release (v0.0.5); some features may still be under refinement.

🚀 Core Features

• Multiprocess Safe: Utilizes multiprocessing.Manager queues to flawlessly support cross-process log collection.
• Asynchronous & Non-blocking: Main and child processes only handle pushing logs into the queue, while disk I/O is managed by a dedicated background listener thread.
• Sequential Integrity: Ensures log record completeness and chronological order during high-concurrency multiprocess writes.
• Lightweight & Easy to Use: Features a modular singleton design, allowing full project log configuration in just a few lines of code.

In the standard logging library, multiple processes writing to the same file simultaneously often lead to:

1. Content Corruption: Race conditions for file locks can result in fragmented or lost log lines.
2. Performance Bottlenecks: Disk I/O is a slow operation; synchronous writes block the main business logic. py-async-log utilizes a Producer-Consumer model to aggregate logs from all processes into a unified queue, which is then written by a single thread in the main process, completely eliminating these issues.

---

📦 Installation

Install directly via pip:

pip install py-async-log

---

🛠️ Quick Start

import logging
import multiprocessing
import os
import time

from py_async_log import get_manager

# Define a temporary log filename
TEST_LOG_FILE = "test_run.log"

def worker_task(config_info, message):
    """Simulate a task in a child process"""
    # Unpack: config_info is a tuple of (initialization_function, arguments)
    config_fn, args = config_info
    config_fn(*args)  # Execute worker-side initialization

    logger = logging.getLogger("test_worker")
    logger.info(message)

def test_multiprocessing_logging():
    """Test if multiple processes can correctly write logs to the same file"""

    # 1. Environment Cleanup: Remove existing log file from previous runs
    if os.path.exists(TEST_LOG_FILE):
        os.remove(TEST_LOG_FILE)

    # 2. Initialize the Manager
    log_mgr = get_manager(TEST_LOG_FILE, level=logging.INFO)
    log_mgr.start_logging()

    try:
        # Get serializable configuration for worker processes
        config_info = log_mgr.get_worker_config_args()

        # 3. Start multiple child processes to write logs simultaneously
        messages = [f"Test message from process {i}" for i in range(3)]
        processes = []

        for msg in messages:
            p = multiprocessing.Process(target=worker_task, args=(config_info, msg))
            p.start()
            processes.append(p)

        for p in processes:
            p.join()

        # 4. Allow a small buffer time for QueueListener to complete disk I/O
        time.sleep(0.5)
        log_mgr.stop_logging()

        # 5. Verify results
        assert os.path.exists(TEST_LOG_FILE), "Log file was not generated"

        with open(TEST_LOG_FILE, "r", encoding="utf-8") as f:
            content = f.read()
            for msg in messages:
                assert msg in content, f"Log missing message: {msg}"
        print("Multiprocessing logging test passed!")

    except Exception as e:
        print(f"An error occurred during testing: {e}")
        raise
    finally:
        # 6. Final cleanup after testing
        if os.path.exists(TEST_LOG_FILE):
            os.remove(TEST_LOG_FILE)

if __name__ == "__main__":
    test_multiprocessing_logging()

---

📜 License

This project is licensed under the Apache License 2.0.