Metadata-Version: 2.4
Name: py-async-log
Version: 0.0.6
Summary: A high-performance multiprocess-safe asynchronous logging tool.一个支持多进程多线程的日志封装库
Author-email: vin <techcn@qq.com>
License: Apache-2.0
Keywords: logging,multiprocessing,async,queue
Classifier: Intended Audience :: Developers
Classifier: Topic :: System :: Logging
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Operating System :: OS Independent
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# py-async-log

[](https://opensource.org/licenses/Apache-2.0)
[](https://www.python.org/)

**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:

```bash
pip install py-async-log
```

---

## 🛠️ Quick Start

```python
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](LICENSE).
