Python
March 9

Простой асинхронный обработчик логирования на Python

В современных высоконагруженных приложениях блокирующие операции вроде синхронного логирования могут стать узким местом. Представленный асинхронный обработчик решает эту проблему, используя очередь задач и пул потоков для немедленной обработки сообщений без остановки основного цикла событий.

Code: Source on github gist

Как это работает

  1. Очередь и потоки
    Сообщения помещаются в очередь Queue, которая обрабатывается в отдельном потоке. Это позволяет избежать блокировки асинхронного кода при записи в файл или выводе в консоль.
  2. Работа с уровнями логирования
    Через перечисление LogLvlEnum поддерживаются стандартные уровни: DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL. Каждому уровню соответствует метод логгера.
  3. Настройка через конфигурацию
    Можно задать путь к файлу, уровень логирования, формат сообщений и даже кастомные обработчики (handlers), как в стандартном модуле logging

Ключевые особенности

  • Неблокирующий ввод/вывод
    Запись логов происходит в фоновом потоке, что сохраняет отзывчивость приложения.
  • Работа с asyncio
    Все методы логирования (debug, info и др.) являются асинхронными и возвращают корутины, совместимые с async/await.
  • Гибкость
    Поддержка вывода как в файл (с режимами перезаписи/вставки), так и в консоль. Форматирование даты и сообщений настраивается через параметры.

Usage:

import async_logging
import asyncio
import logging

# Инициализация логгера с записью в файл
logger = async_logging.AsyncLogger(
    filename="app.log",
    level=logging.DEBUG,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

async def main():
    await logger.info("Приложение запущено")
    try:
        # ... рабочий код ...
    except Exception as e:
        await logger.error(f"Ошибка выполнения: {e}")

asyncio.run(main())

Этот асинхронный обработчик сочетает простоту настройки и эффективность. Он особенно полезен в приложениях с интенсивным вводом-выводом, где синхронное логирование может стать причиной задержек. Исходный код доступен на GitHub для модификации и использования.

Code: Source on github gist