Настройка логирования Python: вывод в консоль и файл одновременно

import logging

logging.basicConfig(level=logging.INFO)

logging.debug("Это отладочное сообщение") # Не отобразится
logging.info("Программа запущена") # Отобразится
logging.warning("Предупреждение") # Отобразится
logging.error("Произошла ошибка") # Отобразится
logging.critical("Критическая ошибка!") # Отобразится

import logging

logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)

logging.debug("Детали операции") # Теперь отобразится с дополнительной информацией

import logging

# Создаем логгер для конкретного модуля
logger = logging.getLogger('my_module')
logger.setLevel(logging.DEBUG)

# Создаем обработчик для вывода в консоль
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# Создаем форматтер
formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
console_handler.setFormatter(formatter)

# Добавляем обработчик к логгеру
logger.addHandler(console_handler)

# Теперь можно логировать
logger.debug("Debug message") # Не отобразится в консоли из-за уровня обработчика
logger.info("Info message") # Отобразится

import logging

logging.basicConfig(
level=logging.INFO,
filename='app.log',
filemode='w',
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s'
)

logging.info("Эта информация будет записана в файл app.log")

import logging
from logging.handlers import RotatingFileHandler

# Создаем логгер
logger = logging.getLogger('file_logger')
logger.setLevel(logging.INFO)

# Создаем обработчик с ротацией файлов
# Максимальный размер файла – 1MB, создаем до 5 файлов
handler = RotatingFileHandler('app.log', maxBytes=1024*1024, backupCount=5)
handler.setLevel(logging.INFO)

# Настраиваем форматирование
formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)

# Добавляем обработчик к логгеру
logger.addHandler(handler)

# Логируем сообщения
logger.info("Это информационное сообщение")
logger.error("Это сообщение об ошибке")

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

# Функция для компрессии старых логов
def compress_log(source, dest):
with open(source, 'rb') as f_in:
with gzip.open(f"{dest}.gz", 'wb') as f_out:
f_out.writelines(f_in)
os.remove(source)

# Класс обработчика с компрессией
class CompressedTimedRotatingFileHandler(TimedRotatingFileHandler):
def doRollover(self):
# Вызываем оригинальный метод ротации
super().doRollover()
# Получаем имя предыдущего файла лога (который только что был ротирован)
if self.backupCount > 0:
for i in range(self.backupCount – 1, 0, -1):
sfn = self.rotation_filename("%s.%d" % (self.baseFilename, i))
dfn = self.rotation_filename("%s.%d" % (self.baseFilename, i + 1))
if os.path.exists(sfn):
if os.path.exists(dfn):
os.remove(dfn)
compress_log(sfn, dfn)

# Создаем логгер с компрессией по времени
logger = logging.getLogger('compressed_logger')
logger.setLevel(logging.INFO)

# Ротация каждый день в полночь, хранение логов за 30 дней
handler = CompressedTimedRotatingFileHandler(
'app.log',
when='midnight',
interval=1,
backupCount=30
)
formatter = logging.Formatter('%(asctime)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Пример использования
logger.info("Важная информация для хранения")

import logging

# Создаем логгер
logger = logging.getLogger('dual_logger')
logger.setLevel(logging.DEBUG) # Устанавливаем базовый уровень для логгера

# Создаем обработчик для консоли
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO) # В консоль выводим от INFO и выше
console_formatter = logging.Formatter('%(levelname)s – %(message)s')
console_handler.setFormatter(console_formatter)

# Создаем обработчик для файла
file_handler = logging.FileHandler('application.log')
file_handler.setLevel(logging.DEBUG) # В файл записываем все сообщения от DEBUG и выше
file_formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
file_handler.setFormatter(file_formatter)

# Добавляем оба обработчика к логгеру
logger.addHandler(console_handler)
logger.addHandler(file_handler)

# Теперь можем логировать сообщения
logger.debug("Это отладочное сообщение (только в файл)")
logger.info("Это информационное сообщение (в консоль и файл)")
logger.warning("Это предупреждение (в консоль и файл)")
logger.error("Это сообщение об ошибке (в консоль и файл)")

import logging
import logging.config

# Словарь с конфигурацией логирования
logging_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'simple': {
'format': '%(levelname)s – %(message)s'
},
'detailed': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(filename)s:%(lineno)d – %(message)s',
'datefmt': '%Y-%m-%d %H:%M:%S'
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'WARNING',
'formatter': 'simple',
'stream': 'ext://sys.stdout',
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'level': 'DEBUG',
'formatter': 'detailed',
'filename': 'app.log',
'maxBytes': 10485760, # 10MB
'backupCount': 3,
},
'errors_file': {
'class': 'logging.FileHandler',
'level': 'ERROR',
'formatter': 'detailed',
'filename': 'errors.log',
'mode': 'a',
},
},
'loggers': {
'': { # Корневой логгер
'handlers': ['console', 'file'],
'level': 'DEBUG',
'propagate': True
},
'critical_module': { # Логгер для критически важного модуля
'handlers': ['console', 'file', 'errors_file'],
'level': 'INFO',
'propagate': False,
},
}
}

# Применяем конфигурацию
logging.config.dictConfig(logging_config)

# Получаем логгеры
root_logger = logging.getLogger()
critical_logger = logging.getLogger('critical_module')

# Пример использования
root_logger.debug("Отладочная информация (только в файл)")
root_logger.warning("Предупреждение (в консоль и файл)")

critical_logger.info("Информация о критическом модуле (все обработчики)")
critical_logger.error("Ошибка в критическом модуле (все обработчики + errors.log)")

import json
import logging.config
import os

def setup_logging(
default_path='logging.json',
default_level=logging.INFO,
env_key='LOG_CFG'
):
"""Настройка логирования из файла конфигурации."""
path = os.getenv(env_key, default_path)
if os.path.exists(path):
with open(path, 'rt') as f:
config = json.load(f)
logging.config.dictConfig(config)
else:
logging.basicConfig(level=default_level)

# Вызываем функцию настройки в начале программы
setup_logging()

# Теперь можно получать настроенные логгеры
logger = logging.getLogger(__name__)
logger.info("Приложение запущено")

import logging

# Создаём логгер с расширенным форматированием
logger = logging.getLogger('advanced_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
formatter = logging.Formatter(
fmt='%(asctime)s.%(msecs)03d [%(processName)s:%(process)d] [%(threadName)s:%(thread)d] '
'%(levelname)s %(name)s.%(funcName)s:%(lineno)d – %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# Пример использования
def process_data():
logger.info("Начинаем обработку данных")
try:
result = 10 / 0 # Вызовет ошибку деления на ноль
except Exception as e:
logger.exception("Ошибка при обработке: %s", e)
logger.debug("Завершаем функцию")

process_data()

import logging
import json
import datetime
import traceback
import sys

class JsonFormatter(logging.Formatter):
"""Форматтер для вывода логов в формате JSON."""

def format(self, record):
log_record = {
"timestamp": datetime.datetime.fromtimestamp(record.created).isoformat(),
"level": record.levelname,
"logger": record.name,
"module": record.module,
"function": record.funcName,
"line": record.lineno,
"message": record.getMessage(),
"process": record.process,
"thread": record.thread
}

if hasattr(record, 'request_id'):
log_record["request_id"] = record.request_id

if record.exc_info:
log_record["exception"] = {
"type": record.exc_info[0].__name__,
"message": str(record.exc_info[1]),
"traceback": traceback.format_exception(*record.exc_info)
}

return json.dumps(log_record)

# Настраиваем логгер с JSON-форматированием
logger = logging.getLogger('json_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)

# Добавляем дополнительную информацию в логи
class RequestIdFilter(logging.Filter):
"""Фильтр для добавления ID запроса в записи логов."""

def __init__(self, request_id):
super().__init__()
self.request_id = request_id

def filter(self, record):
record.request_id = self.request_id
return True

# Пример использования
def process_request(request_id):
# Добавляем фильтр с ID запроса
request_filter = RequestIdFilter(request_id)
logger.addFilter(request_filter)

logger.info("Обработка запроса")
try:
# Имитация ошибки
if request_id % 2 == 0:
raise ValueError("Некорректный запрос")
logger.info("Запрос успешно обработан")
except Exception as e:
logger.exception("Ошибка при обработке запроса")
finally:
# Удаляем фильтр после завершения
logger.removeFilter(request_filter)

# Имитируем обработку нескольких запросов
process_request(1)
process_request(2)

import logging
import logging.handlers
import queue
import threading

# Создаём очередь для логов
log_queue = queue.Queue(-1) # Неограниченная очередь

# Настраиваем обработчик с очередью
queue_handler = logging.handlers.QueueHandler(log_queue)

# Основной логгер приложения
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
root_logger.addHandler(queue_handler)

# Настраиваем слушатели, которые будут обрабатывать сообщения из очереди
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter('%(asctime)s – %(levelname)s – %(message)s'))

file_handler = logging.FileHandler('app.log')
file_handler.setFormatter(logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s'))

# Создаём и запускаем слушателя очереди в отдельном потоке
listener = logging.handlers.QueueListener(
log_queue, console_handler, file_handler, respect_handler_level=True
)
listener.start()

# Пример использования
logger = logging.getLogger(__name__)
logger.info("Приложение запущено")

# Имитация интенсивного логирования
for i in range(1000):
logger.debug(f"Итерация {i}")

logger.info("Работа завершена")

# При завершении приложения
listener.stop()

import logging
import sys
import socket
from logging.handlers import SocketHandler, WatchedFileHandler

class FallbackHandler(logging.Handler):
"""Обработчик с механизмом отказоустойчивости."""

def __init__(self, primary_handler, fallback_handler):
super().__init__()
self.primary_handler = primary_handler
self.fallback_handler = fallback_handler

def emit(self, record):
try:
self.primary_handler.emit(record)
except Exception:
try:
self.fallback_handler.emit(record)
except Exception:
# Если оба обработчика отказали, выводим в stderr
sys.stderr.write(f"КРИТИЧЕСКАЯ ОШИБКА ЛОГИРОВАНИЯ: {record.getMessage()}\n")

def close(self):
self.primary_handler.close()
self.fallback_handler.close()
super().close()

# Настраиваем основные и резервные обработчики
try:
# Основной обработчик – удалённый лог-сервер
remote_handler = SocketHandler('logserver.example.com', 9999)

# Резервный обработчик – локальный файл
local_handler = WatchedFileHandler('fallback.log')

# Объединяем их в отказоустойчивый обработчик
handler = FallbackHandler(remote_handler, local_handler)

# Настраиваем корневой логгер
root_logger = logging.getLogger()
root_logger.setLevel(logging.INFO)
root_logger.addHandler(handler)

# Сообщаем о успешной настройке
root_logger.info("Система логирования инициализирована")

except Exception as e:
# Если не удалось настроить логирование, используем базовую конфигурацию
logging.basicConfig(
level=logging.WARNING,
format='%(asctime)s – %(levelname)s – %(message)s',
filename='emergency.log'
)
logging.error(f"Ошибка настройки логирования: {e}")