Python logging: полный обзор инструментов для отладки и мониторинга

Расширенные примеры работы с логированием

Пример 1: Кастомный обработчик с отправкой логов по сети (SysLogHandler)

import logging
from logging.handlers import SysLogHandler

logger = logging.getLogger('syslog_logger')
logger.setLevel(logging.DEBUG)

handler = SysLogHandler(address='/dev/log')
handler.setFormatter(logging.Formatter('%(name)s: %(message)s'))
logger.addHandler(handler)

logger.critical('Критическая ошибка системы')

Результат (в системном журнале): сообщение появляется с facility user (по умолчанию). Для изменения facility передаётся facility аргумент.

Пример 2: Использование MemoryHandler для буферизации и отправки пакетов

import logging
from logging.handlers import MemoryHandler, RotatingFileHandler

log_file = 'buffered.log'
file_handler = RotatingFileHandler(log_file, maxBytes=1024, backupCount=1)
file_handler.setLevel(logging.ERROR)

memory_handler = MemoryHandler(capacity=10, target=file_handler, flushLevel=logging.ERROR)

logger = logging.getLogger('memory_logger')
logger.setLevel(logging.DEBUG)
logger.addHandler(memory_handler)

# Эти сообщения накапливаются в памяти
for i in range(5):
    logger.info(f'Сообщение {i}')

# При ошибке все накопленные записи сбрасываются в файл
logger.error('Ошибка, вызывающая сброс')

Содержимое buffered.log (пример):
INFO:memory_logger:Сообщение 0
INFO:memory_logger:Сообщение 1
INFO:memory_logger:Сообщение 2
INFO:memory_logger:Сообщение 3
INFO:memory_logger:Сообщение 4
ERROR:memory_logger:Ошибка, вызывающая сброс

Пояснение: MemoryHandler буферизирует сообщения до тех пор, пока не будет достигнут лимит capacity или не поступит запись уровня flushLevel (по умолчанию ERROR). Затем все записи переносятся в целевой обработчик.

Пример 3: Фильтры для выборочного логирования

import logging

class ImportantFilter(logging.Filter):
    def filter(self, record):
        return '[ВАЖНО]' in record.getMessage()

logger = logging.getLogger('filter_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))
handler.addFilter(ImportantFilter())
logger.addHandler(handler)

logger.info('Обычное сообщение')          # не попадёт в вывод
logger.info('[ВАЖНО] Срочное уведомление') # попадёт

Результат:

[ВАЖНО] Срочное уведомление

Фильтры можно добавлять как к логгерам (через addFilter), так и к обработчикам.

Пример 4: Настройка логирования для многопоточных приложений

import logging
import threading
import time

logging.basicConfig(
    level=logging.DEBUG,
    format='[%(threadName)s] %(message)s'
)
logger = logging.getLogger('thread_logger')

def worker():
    for i in range(3):
        logger.debug(f'Шаг {i}')
        time.sleep(0.1)

threads = []
for _ in range(2):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

Результат (порядок может отличаться):

[Thread-1] Шаг 0
[Thread-2] Шаг 0
[Thread-1] Шаг 1
[Thread-2] Шаг 1
[Thread-1] Шаг 2
[Thread-2] Шаг 2

Библиотека logging потокобезопасна, но при использовании одного файлового обработчика может возникать конкуренция. Для избежания потери записей в многопоточных сценариях используют QueueHandler и отдельный поток с QueueListener (доступно с Python 3.2+).

Пример 5: Использование QueueListener и QueueHandler для асинхронного логирования

import logging
from logging.handlers import QueueHandler, QueueListener, RotatingFileHandler
from queue import Queue

# Создаём очередь и хендлер
transfer_queue = Queue()

# Обработчик, который будет фактически выполнять запись
file_handler = RotatingFileHandler('async.log', maxBytes=1024, backupCount=1)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s'))

# Слушатель очереди
listener = QueueListener(transfer_queue, file_handler)
listener.start()

# Логгер использует QueueHandler
logger = logging.getLogger('async_logger')
logger.setLevel(logging.DEBUG)
queue_handler = QueueHandler(transfer_queue)
logger.addHandler(queue_handler)

# Теперь вызовы logger.debug() не блокируются на запись в файл
for i in range(10):
    logger.debug(f'Асинхронное сообщение {i}')

# В конце программы остановить listener
listener.stop()

Пояснение: QueueListener работает в отдельном потоке, вычитывает записи из очереди и передаёт их целевому обработчику. Это повышает производительность при интенсивной записи.

Пример 6: Логирование с помощью dictConfig из JSON-файла

import json
import logging.config

config_str = '''
{
    "version": 1,
    "formatters": {
        "json_formatter": {
            "format": "{\"time\": \"%(asctime)s\", \"level\": \"%(levelname)s\", \"message\": \"%(message)s\"}"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "DEBUG",
            "formatter": "json_formatter"
        }
    },
    "loggers": {
        "json_logger": {
            "handlers": ["console"],
            "level": "DEBUG",
            "propagate": false
        }
    }
}
'''

config = json.loads(config_str)
logging.config.dictConfig(config)

logger = logging.getLogger('json_logger')
logger.info('Это сообщение в формате JSON')

{"time": "2025-03-15 14:22:01,234", "level": "INFO", "message": "Это сообщение в формате JSON"}

Такой подход удобен при интеграции с системами сбора логов (ELK, Splunk и т.д.).