RotatingFileHandler, TimedRotatingFileHandler, MemoryHandler, SMTPHandler
Handlers определяют, куда отправляются логи. Выбор правильного handler критичен для production-логирования.
Проблема: Файл лога растёт бесконечно, занимая всё место на диске.
Решение: RotatingFileHandler ротирует файл при достижении размера.
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. 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
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:
Проблема: Нужно ротировать логи по расписанию (каждый день, неделю).
Решение: TimedRotatingFileHandler ротирует по времени.
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)| Значение | Интервал | Пример |
|---|---|---|
'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 марта
Важно: TimedRotatingFileHandler добавляет дату к имени файла в формате .YYYY-MM-DD.
Проблема: 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
)Проблема: Частая запись в файл/сеть снижает производительность.
Решение: MemoryHandler буферизует записи и отправляет пачкой.
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()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
Проблема: Нужно отправлять 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 записями + это сообщениеПроблема: Нужно получать уведомления о критических ошибках.
Решение: SMTPHandler отправляет логи по email.
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)smtp_handler = SMTPHandler(
mailhost=('smtp.example.com', 587),
fromaddr='app@example.com',
toaddrs=['oncall@example.com'],
subject='Application Error',
credentials=('username', 'password'),
secure=() # TLS
)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'
)Проблема: 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
}Проблема: Нужно централизованное логирование для нескольких серверов.
Решение: SocketHandler отправляет логи на центральный сервер.
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)# 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()Проблема: Нужно выводить логи в несколько мест с разными настройками.
Решение: Несколько 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()# ✅ Хорошо
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('app.log', maxBytes=100*1024*1024, backupCount=10)
# ❌ Плохо: файл растёт бесконечно
handler = logging.FileHandler('app.log')# ✅ Хорошо: откладывает открытие файла
handler = RotatingFileHandler('app.log', delay=True)
# ❌ Плохо: файл открывается немедленно
handler = RotatingFileHandler('app.log')# ✅ Хорошо для Unix с logrotate
from logging.handlers import WatchedFileHandler
handler = WatchedFileHandler('/var/log/app/app.log')# ✅ Хорошо
smtp_handler.setLevel(logging.ERROR)
# ❌ Плохо: спам email'ами
smtp_handler.setLevel(logging.INFO)# ✅ Хорошо: буферизация для частых записей
memory_handler = MemoryHandler(capacity=100, target=file_handler)
# ❌ Плохо: прямая запись для каждого лога
logger.addHandler(file_handler) # без буферизацииВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.