Продвинутые Handlers — Python Logging: от основ до

Продвинутые Handlers в Python logging

Handlers определяют, куда отправляются логи. Выбор правильного handler критичен для production-логирования.

#1. RotatingFileHandler — ротация по размеру

Проблема: Файл лога растёт бесконечно, занимая всё место на диске.

Решение: RotatingFileHandler ротирует файл при достижении размера.

#1.1. Базовое использование

from logging.handlers import RotatingFileHandler
import logging

# Ротация при 10MB, хранение 5 backup файлов
handler = RotatingFileHandler(
    'app.log',
    maxBytes=10 * 1024 * 1024,  # 10 MB
    backupCount=5
)

logger = logging.getLogger(__name__)
logger.addHandler(handler)

Структура файлов:

app.log       ← текущий файл (пишется сейчас)
app.log.1     ← последний backup (самый новый)
app.log.2     ← предыдущий backup
app.log.3     ← ...
app.log.4     ← ...
app.log.5     ← самый старый backup (будет удалён при следующей ротации)

#1.2. Как работает ротация

1. app.log достигает 10MB
2. app.log.5 удаляется (если существует)
3. app.log.4 → app.log.5
4. app.log.3 → app.log.4
5. app.log.2 → app.log.3
6. app.log.1 → app.log.2
7. app.log → app.log.1
8. Создаётся новый app.log

#1.3. Production конфигурация

from logging.handlers import RotatingFileHandler
import logging

def setup_rotating_handler(filename: str, 
                           max_mb: int = 100, 
                           backup_count: int = 10) -> RotatingFileHandler:
    """Настройка RotatingFileHandler для production."""
    handler = RotatingFileHandler(
        filename,
        maxBytes=max_mb * 1024 * 1024,
        backupCount=backup_count,
        encoding='utf-8',
        delay=True  # Откладывает открытие файла до первой записи
    )
    
    # Formatter
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    handler.setLevel(logging.DEBUG)
    
    return handler

# Использование
logger = logging.getLogger(__name__)
logger.addHandler(setup_rotating_handler('/var/log/app/app.log'))

Параметр delay=True:

Откладывает открытие файла до первой записи лога
Полезно, если приложение форкается (например, uWSGI, Gunicorn)
Избегает проблем с правами доступа при запуске от разных пользователей

#2. TimedRotatingFileHandler — ротация по времени

Проблема: Нужно ротировать логи по расписанию (каждый день, неделю).

Решение: TimedRotatingFileHandler ротирует по времени.

#2.1. Базовое использование

from logging.handlers import TimedRotatingFileHandler
import logging

# Ротация каждый день в полночь, хранение 30 дней
handler = TimedRotatingFileHandler(
    'app.log',
    when='midnight',
    interval=1,
    backupCount=30,
    encoding='utf-8'
)

logger = logging.getLogger(__name__)
logger.addHandler(handler)

#2.2. Параметр when

Значение	Интервал	Пример
`'S'`	Секунды	`when='S', interval=60` — каждые 60 секунд
`'M'`	Минуты	`when='M', interval=15` — каждые 15 минут
`'H'`	Часы	`when='H', interval=6` — каждые 6 часов
`'D'`	Дни	`when='D', interval=1` — каждый день
`'W0'-'W6'`	Дни недели	`when='W0'` — каждый понедельник
`'midnight'`	Полночь	`when='midnight'` — каждый день в 00:00

#2.3. Структура файлов

app.log              ← текущий файл
app.log.2026-03-17   ← backup за 17 марта
app.log.2026-03-16   ← backup за 16 марта
app.log.2026-03-15   ← backup за 15 марта

Важно: TimedRotatingFileHandler добавляет дату к имени файла в формате .YYYY-MM-DD.

#2.4. Сжатие старых файлов

Проблема: Backup файлы занимают много места.

Решение: Кастомный handler со сжатием.

from logging.handlers import TimedRotatingFileHandler
import gzip
import shutil
import os
from datetime import datetime

class CompressingTimedRotatingFileHandler(TimedRotatingFileHandler):
    def doRollover(self):
        super().doRollover()
        
        # Сжатие старого файла
        old_file = self.stream.name + '.1'
        if os.path.exists(old_file):
            compressed_file = old_file + '.gz'
            with open(old_file, 'rb') as f_in:
                with gzip.open(compressed_file, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            os.remove(old_file)

# Использование
handler = CompressingTimedRotatingFileHandler(
    'app.log',
    when='midnight',
    backupCount=30
)

#3. MemoryHandler — буферизация в памяти

Проблема: Частая запись в файл/сеть снижает производительность.

Решение: MemoryHandler буферизует записи и отправляет пачкой.

#3.1. Базовое использование

from logging.handlers import MemoryHandler, RotatingFileHandler
import logging

# Целевой handler (куда отправлять буфер)
target_handler = RotatingFileHandler('app.log', maxBytes=10*1024*1024)

# MemoryHandler буферизует 100 записей
memory_handler = MemoryHandler(
    capacity=100,           # Размер буфера
    flushLevel=logging.ERROR,  # Уровень для принудительного flush
    target=target_handler   # Куда отправлять при flush
)

logger = logging.getLogger(__name__)
logger.addHandler(memory_handler)

# Логи буферизуются
logger.info("Info message")      # В буфере
logger.debug("Debug message")    # В буфере

# При ERROR — автоматический flush
logger.error("Error occurred")   # Flush буфера + это сообщение

# Ручной flush
memory_handler.flush()

#3.2. Как работает MemoryHandler

1. logger.info() → LogRecord попадает в буфер MemoryHandler
2. Буфер заполняется (capacity=100) → автоматический flush в target
3. logger.error() → flushLevel=ERROR → немедленный flush
4. memory_handler.flush() → ручной flush
5. При flush: все записи из буфера отправляются в target_handler

#3.3. Use case: отправка email при критических ошибках

Проблема: Нужно отправлять email только при CRITICAL ошибках с историей.

Решение: MemoryHandler + SMTPHandler.

from logging.handlers import MemoryHandler, SMTPHandler
import logging

# SMTPHandler отправляет email
smtp_handler = SMTPHandler(
    mailhost=('smtp.example.com', 587),
    fromaddr='app@example.com',
    toaddrs=['oncall@example.com'],
    subject='Critical Error in Application'
)
smtp_handler.setLevel(logging.ERROR)

# MemoryHandler буферизует ERROR+ логи
memory_handler = MemoryHandler(
    capacity=10,
    flushLevel=logging.CRITICAL,
    target=smtp_handler
)
memory_handler.setLevel(logging.ERROR)

logger = logging.getLogger(__name__)
logger.addHandler(memory_handler)

# ERROR логи буферизуются
logger.error("First error")
logger.error("Second error")

# CRITICAL лог вызывает flush и отправку email
logger.critical("System failure")
# → Отправляет email со всеми buffered записями + это сообщение

#4. SMTPHandler — отправка логов по email

Проблема: Нужно получать уведомления о критических ошибках.

Решение: SMTPHandler отправляет логи по email.

#4.1. Базовое использование

from logging.handlers import SMTPHandler
import logging

# Email при ERROR и выше
smtp_handler = SMTPHandler(
    mailhost=('smtp.example.com', 587),
    fromaddr='app@example.com',
    toaddrs=['admin@example.com', 'oncall@example.com'],
    subject='Application Error'
)
smtp_handler.setLevel(logging.ERROR)

logger = logging.getLogger(__name__)
logger.addHandler(smtp_handler)

#4.2. С аутентификацией и TLS

smtp_handler = SMTPHandler(
    mailhost=('smtp.example.com', 587),
    fromaddr='app@example.com',
    toaddrs=['oncall@example.com'],
    subject='Application Error',
    credentials=('username', 'password'),
    secure=()  # TLS
)

#4.3. С кастомным subject (имя хоста, уровень)

import logging
import socket
from logging.handlers import SMTPHandler

class CustomSMTPHandler(SMTPHandler):
    def getSubject(self, record):
        prefix = super().getSubject(record)
        hostname = socket.gethostname()
        return f"{prefix} - {hostname} - {record.levelname}"

smtp_handler = CustomSMTPHandler(
    mailhost=('smtp.example.com', 587),
    fromaddr='app@example.com',
    toaddrs=['oncall@example.com'],
    subject='Application Error'
)

#5. WatchedFileHandler — мониторинг внешнего rotation

Проблема: logrotate перемещает файл лога, но Python продолжает писать в старый файл.

Решение: WatchedFileHandler проверяет inode и reopening при изменении.

from logging.handlers import WatchedFileHandler
import logging

# WatchedFileHandler следит за файлом
handler = WatchedFileHandler('/var/log/app/app.log')

logger = logging.getLogger(__name__)
logger.addHandler(handler)

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

1. Приложение пишет в app.log (inode 12345)
2. logrotate перемещает app.log → app.log.1 (inode 12345)
3. logrotate создаёт новый app.log (inode 67890)
4. WatchedFileHandler обнаруживает изменение inode
5. Handler автоматически открывает новый app.log (inode 67890)

Использование с logrotate:

# /etc/logrotate.d/app
/var/log/app/app.log {
    daily
    rotate 30
    compress
    delaycompress
    missingok
    notifempty
    create 0644 app app
}

#6. SocketHandler — отправка логов по сети

Проблема: Нужно централизованное логирование для нескольких серверов.

Решение: SocketHandler отправляет логи на центральный сервер.

#6.1. Клиент (отправка)

from logging.handlers import SocketHandler
import logging

# Отправка на центральный лог-сервер
socket_handler = SocketHandler(
    host='log-server.example.com',
    port=9020
)

logger = logging.getLogger(__name__)
logger.addHandler(socket_handler)

#6.2. Сервер (приём)

# log_server.py
import logging
import logging.handlers
import socketserver
import struct

class LogRecordStreamHandler(socketserver.StreamRequestHandler):
    def handle(self):
        while True:
            chunk = self.rfile.read(4)
            if not chunk:
                break
            slen = struct.unpack('>L', chunk)[0]
            chunk = self.rfile.read(slen)
            log_record = logging.handlers.makeLogRecord(chunk)
            
            logger = logging.getLogger(log_record.name)
            logger.handle(log_record)

class LogRecordSocketReceiver(socketserver.ThreadingTCPServer):
    allow_reuse_address = True

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    server = LogRecordSocketReceiver(('0.0.0.0', 9020), LogRecordStreamHandler)
    print('Starting log server on port 9020...')
    server.serve_forever()

#7. Комбинация handlers

Проблема: Нужно выводить логи в несколько мест с разными настройками.

Решение: Несколько handlers для разных целей.

import logging
from logging.handlers import (
    RotatingFileHandler,
    TimedRotatingFileHandler,
    SMTPHandler,
    MemoryHandler
)

def setup_production_logging():
    logger = logging.getLogger('app')
    logger.setLevel(logging.DEBUG)
    
    # 1. Консоль — INFO и выше, кратко
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
    
    # 2. Файл — DEBUG и выше, подробно, ротация по размеру
    file_handler = RotatingFileHandler(
        '/var/log/app/app.log',
        maxBytes=100*1024*1024,
        backupCount=10
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s - %(filename)s:%(lineno)d'
    ))
    
    # 3. Audit лог — только важные события, ротация по времени
    audit_handler = TimedRotatingFileHandler(
        '/var/log/app/audit.log',
        when='midnight',
        backupCount=365  # Хранить год
    )
    audit_handler.setLevel(logging.INFO)
    audit_handler.addFilter(lambda record: record.name.startswith('app.audit'))
    
    # 4. Email — CRITICAL ошибки
    smtp_handler = SMTPHandler(
        mailhost=('smtp.example.com', 587),
        fromaddr='app@example.com',
        toaddrs=['oncall@example.com'],
        subject='CRITICAL Error',
        credentials=('user', 'pass'),
        secure=()
    )
    smtp_handler.setLevel(logging.CRITICAL)
    
    # 5. Буфер для ERROR логов (отправка пачкой)
    error_buffer = MemoryHandler(
        capacity=50,
        flushLevel=logging.CRITICAL
    )
    error_buffer.setLevel(logging.ERROR)
    error_buffer.setTarget(file_handler)
    
    # Добавление handlers
    logger.addHandler(console)
    logger.addHandler(file_handler)
    logger.addHandler(audit_handler)
    logger.addHandler(smtp_handler)
    logger.addHandler(error_buffer)
    
    return logger

# Использование
logger = setup_production_logging()

#8. Best Practices

#8.1. Используйте RotatingFileHandler для production

# ✅ Хорошо
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('app.log', maxBytes=100*1024*1024, backupCount=10)

# ❌ Плохо: файл растёт бесконечно
handler = logging.FileHandler('app.log')

#8.2. Включайте delay=True

# ✅ Хорошо: откладывает открытие файла
handler = RotatingFileHandler('app.log', delay=True)

# ❌ Плохо: файл открывается немедленно
handler = RotatingFileHandler('app.log')

#8.3. Используйте WatchedFileHandler с logrotate

# ✅ Хорошо для Unix с logrotate
from logging.handlers import WatchedFileHandler
handler = WatchedFileHandler('/var/log/app/app.log')

#8.4. SMTPHandler только для ERROR+

# ✅ Хорошо
smtp_handler.setLevel(logging.ERROR)

# ❌ Плохо: спам email'ами
smtp_handler.setLevel(logging.INFO)

#8.5. MemoryHandler для производительности

# ✅ Хорошо: буферизация для частых записей
memory_handler = MemoryHandler(capacity=100, target=file_handler)

# ❌ Плохо: прямая запись для каждого лога
logger.addHandler(file_handler)  # без буферизации

app.log ← текущий файл (пишется сейчас) app.log.1 ← последний backup (самый новый) app.log.2 ← предыдущий backup app.log.3 ← ... app.log.4 ← ... app.log.5 ← самый старый backup (будет удалён при следующей ротации)

1. app.log достигает 10MB 2. app.log.5 удаляется (если существует) 3. app.log.4 → app.log.5 4. app.log.3 → app.log.4 5. app.log.2 → app.log.3 6. app.log.1 → app.log.2 7. app.log → app.log.1 8. Создаётся новый app.log

Значение

Интервал

Пример

'S'

Секунды

when='S', interval=60 — каждые 60 секунд

'M'

Минуты

when='M', interval=15 — каждые 15 минут

'H'

Часы

when='H', interval=6 — каждые 6 часов

'D'

Дни

when='D', interval=1 — каждый день

'W0'-'W6'

Дни недели

when='W0' — каждый понедельник

'midnight'

Полночь

when='midnight' — каждый день в 00:00

app.log ← текущий файл app.log.2026-03-17 ← backup за 17 марта app.log.2026-03-16 ← backup за 16 марта app.log.2026-03-15 ← backup за 15 марта

1. logger.info() → LogRecord попадает в буфер MemoryHandler 2. Буфер заполняется (capacity=100) → автоматический flush в target 3. logger.error() → flushLevel=ERROR → немедленный flush 4. memory_handler.flush() → ручной flush 5. При flush: все записи из буфера отправляются в target_handler

1. Приложение пишет в app.log (inode 12345) 2. logrotate перемещает app.log → app.log.1 (inode 12345) 3. logrotate создаёт новый app.log (inode 67890) 4. WatchedFileHandler обнаруживает изменение inode 5. Handler автоматически открывает новый app.log (inode 67890)