Async Safe Logger

中文文档

A simple, thread-safe, and asyncio-friendly logging wrapper built on top of Python's standard logging module.

AsyncSafeLogger moves log formatting, file I/O, log rotation, retention cleanup, and console output to a dedicated background thread using QueueHandler and QueueListener.

Your application code continues to use Python's standard logging API:

import logging

logger = logging.getLogger(__name__)

logger.info("Hello, world!")

No await is required for normal logging calls.

---

Features

• Built on Python's standard logging module
• Asyncio-friendly — logging calls do not perform file or console I/O on the event-loop thread
• Thread-safe — safe to use from multiple threads and coroutines
• Background logging — formatting and output are handled by a dedicated QueueListener thread
• Automatic log rotation — rotate log files when they reach the configured size
• Log retention cleanup — automatically remove expired log files
• Console and file output — independently configurable
• Multiple independent logger instances — no required global mutable state
• Synchronous and asynchronous lifecycle APIs
• Context manager support
• Zero runtime dependencies — only Python's standard library is required

---

Requirements

• Python 3.13+

The library currently has no third-party runtime dependencies.

---

Installation

Using pip

pip install async-safe-logger

Using uv

If you use uv, add the package to your project with:

uv add async-safe-logger

Or install it directly into the current environment:

uv pip install async-safe-logger

---

Quick Start

The simplest usage is to configure the logger and continue using Python's standard logging API.

import logging

from async_safe_logger import setup_logging, stop_logging

setup_logging()

logger = logging.getLogger(__name__)

logger.info("Hello, world!")
logger.warning("Something may be wrong.")
logger.error("Something went wrong.")

stop_logging()

By default, logs are:

• written to logs/app.log
• printed to stdout
• rotated when the file reaches 2 MB
• retained according to the configured retention policy

---

Asyncio Usage

AsyncSafeLogger is designed for applications using asyncio.

Normal logging calls remain synchronous from the caller's perspective:

import asyncio
import logging

from async_safe_logger import setup_logging, stop_logging_async


async def main():
    setup_logging()

    logger = logging.getLogger(__name__)

    logger.info("Running inside asyncio")
    logger.warning("This logging call does not write directly to disk.")

    try:
        ...
    finally:
        await stop_logging_async()


asyncio.run(main())

Why stop_logging_async()?

Stopping a QueueListener may wait for its background thread to finish processing queued log records.

The synchronous:

stop_logging()

may therefore block while the listener shuts down.

When running inside an asyncio event loop, prefer:

await stop_logging_async()

This moves the potentially blocking shutdown operation to a worker thread.

---

How It Works

The library uses Python's standard QueueHandler and QueueListener.

The normal logging path looks approximately like this:

Application
    │
    │ logger.info(...)
    ▼
logging.Logger
    │
    ▼
QueueHandler
    │
    │ put LogRecord into memory queue
    ▼
queue.Queue
    │
    ▼
QueueListener
    │
    │ background thread
    ▼
┌───────────────┬────────────────┐
│ FileHandler   │ ConsoleHandler │
└───────────────┴────────────────┘
    │                    │
    ▼                    ▼
 app.log              stdout

The important point is that the application thread or asyncio event-loop thread does not directly perform:

• file writes
• console writes
• log rotation
• expired-log cleanup

Those operations are handled by the background listener thread.

This keeps the normal logging path lightweight and avoids performing blocking output operations directly from an asyncio event loop.

---

Configuration

The default logger can be configured through setup_logging().

For example:

from async_safe_logger import setup_logging

setup_logging(
    log_dir="logs",
    log_filename="app.log",
    level=logging.INFO,
    max_bytes=2 * 1024 * 1024,
    backup_count=20,
    retention_days=5,
    file=True,
    console=True,
)

Configuration Options

Parameter	Default	Description
log_dir	"logs"	Directory used for log files
log_filename	"app.log"	Log file name
level	logging.INFO	Minimum logging level
max_bytes	2 MB	Maximum size of a log file before rotation
backup_count	20	Number of rotated log files to keep
retention_days	5	Delete log files older than this number of days
file	True	Enable file logging
console	True	Enable console logging
fmt	See below	Log message format
datefmt	"%Y-%m-%d %H:%M:%S"	Timestamp format

Default format:

%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d - %(message)s

For example:

2026-09-06 20:30:15 | INFO     | myapp.main:main:42 - Server started

---

File and Console Output

File and console logging can be independently enabled.

File only

AsyncSafeLogger(
    file=True,
    console=False,
)

Console only

AsyncSafeLogger(
    file=False,
    console=True,
)

File and console

AsyncSafeLogger(
    file=True,
    console=True,
)

This is also the default configuration.

For safety, the following configuration is rejected:

AsyncSafeLogger(
    file=False,
    console=False,
)

It raises ValueError instead of silently discarding all log messages.

---

Multiple Independent Logger Systems

You can create multiple AsyncSafeLogger instances when different subsystems need different configurations.

For example:

import logging

from async_safe_logger import AsyncSafeLogger


business_logger = AsyncSafeLogger(
    log_dir="logs/business",
    log_filename="business.log",
    level=logging.INFO,
)

debug_logger = AsyncSafeLogger(
    log_dir="logs/debug",
    log_filename="debug.log",
    level=logging.DEBUG,
)

business_logger.setup()
debug_logger.setup()

logging.getLogger("business").info("Business event")
logging.getLogger("debug").debug("Debug information")

# ...

business_logger.stop()
debug_logger.stop()

Each instance owns its own queue, listener, and handlers.

---

Context Manager

AsyncSafeLogger supports both synchronous and asynchronous context managers.

Synchronous

import logging

from async_safe_logger import AsyncSafeLogger


with AsyncSafeLogger() as logger:
    logging.info("Inside the context manager")

The logger is automatically stopped when leaving the with block.

Asynchronous

import logging

from async_safe_logger import AsyncSafeLogger


async with AsyncSafeLogger() as logger:
    logging.info("Inside the async context manager")

When leaving the async with block, the logger is shut down asynchronously.

This is the recommended pattern when the surrounding application is already asynchronous.

---

Standard Logging Compatibility

The main goal of this library is to keep application code compatible with Python's standard logging API.

You can continue to use:

import logging

logger = logging.getLogger(__name__)

logger.debug("Debug message")
logger.info("Information")
logger.warning("Warning")
logger.error("Error")
logger.critical("Critical error")

There is no need to replace your application's logging calls with a custom API such as:

await logger.info(...)

Instead, logging remains familiar:

logger.info(...)

while the output pipeline is handled asynchronously in the background.

---

Log Rotation and Retention

The file handler combines size-based rotation with automatic retention cleanup.

For example:

AsyncSafeLogger(
    log_dir="logs",
    log_filename="app.log",
    max_bytes=2 * 1024 * 1024,
    backup_count=20,
    retention_days=5,
)

When app.log reaches the configured size, it is rotated according to RotatingFileHandler.

After rotation, log files older than the configured retention period are removed.

---

Graceful Shutdown

It is important to stop the logger before the application exits so that queued log records have an opportunity to be processed.

For synchronous applications:

stop_logging()

For asyncio applications:

await stop_logging_async()

A typical asyncio application should use:

async def main():
    setup_logging()

    try:
        ...
    finally:
        await stop_logging_async()

This is especially important for applications that produce logs immediately before shutdown.

---

Design Goals

The project intentionally focuses on a small set of goals:

1. Keep the standard logging API

No custom logging syntax is required.

logger.info("hello")

remains the primary interface.

2. Keep blocking output away from the event loop

The application produces a LogRecord, which is placed into an in-memory queue.

The listener thread performs the actual output work.

3. Minimize runtime dependencies

The current implementation uses only Python's standard library.

4. Support both synchronous and asynchronous applications

The same logger can be used by:

• normal synchronous applications
• asyncio applications
• applications using both threads and asyncio

5. Keep the implementation simple

The library intentionally builds on well-tested components already provided by Python:

• logging
• QueueHandler
• QueueListener
• RotatingFileHandler
• queue.Queue
• threading

Rather than implementing a completely new logging framework.

---

Important Notes

Logging is not completely "non-blocking"

AsyncSafeLogger moves output operations away from the calling thread, but the initial logging operation still performs normal Python logging work and enqueues a LogRecord.

Therefore, it should be understood as:

asyncio-friendly / async-safe logging output

rather than a mathematically guaranteed zero-cost or lock-free logging implementation.

Queue Backpressure

The internal queue is currently unbounded.

This means normal logging calls do not wait for the background listener to consume records.

However, an application that continuously produces logs significantly faster than the output destination can consume them may accumulate an increasing number of LogRecord objects in memory.

For high-volume logging workloads, monitor queue growth and application memory usage.

Shutdown

Always stop the logger during application shutdown:

await stop_logging_async()

for asyncio applications, or:

stop_logging()

for synchronous applications.

---

Project Status

This project is currently in an early development stage.

API design and implementation details may change before the 1.0.0 release.

The current version is:

0.1.0

---

License

MIT License.

See LICENSE for details.