Исходник Простой асинхронный обработчик логирования на Python | async logging

Начинающий
Начинающий
Статус
Оффлайн
Регистрация
15 Фев 2025
Сообщения
27
Реакции
2

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

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

Code:
Пожалуйста, авторизуйтесь для просмотра ссылки.


Как это работает
  1. Очередь и потоки
    Сообщения помещаются в очередь Queue, которая обрабатывается в отдельном потоке. Это позволяет избежать блокировки асинхронного кода при записи в файл или выводе в консоль.
  2. Работа с уровнями логирования
    Через перечисление LogLvlEnum поддерживаются стандартные уровни: DEBUG, INFO, WARNING, ERROR, EXCEPTION, CRITICAL. Каждому уровню соответствует метод логгера.
  3. Настройка через конфигурацию
    Можно задать путь к файлу, уровень логирования, формат сообщений и даже кастомные обработчики (handlers), как в стандартном модуле logging
Ключевые особенности
  • Неблокирующий ввод/вывод
    Запись логов происходит в фоновом потоке, что сохраняет отзывчивость приложения.
  • Работа с asyncio
    Все методы логирования (debug, info и др.) являются асинхронными и возвращают корутины, совместимые с async/await.
  • Гибкость
    Поддержка вывода как в файл (с режимами перезаписи/вставки), так и в консоль. Форматирование даты и сообщений настраивается через параметры.
Использование:
Usage:
Expand Collapse Copy
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())

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