Эффективное логирование в Python: лучшие практики и настройки

import logging

# Настройка базового логгера
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
filename='app.log'
)

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

import logging

# Создание именованного логгера
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

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

# Создание обработчика для записи в файл
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.WARNING)

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

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

# Использование
logger.debug("Отладочная информация")
logger.info("Информационное сообщение")
logger.warning("Предупреждение")

import logging

logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler("debug.log"),
logging.StreamHandler()
]
)

logger = logging.getLogger(__name__)

import logging
from logging.handlers import RotatingFileHandler

logger = logging.getLogger('my_app')
logger.setLevel(logging.INFO)

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

formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)

logger.addHandler(handler)

# Использование
for i in range(10000):
logger.info(f"Это информационное сообщение #{i}")

import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
def format(self, record):
log_record = {
"timestamp": datetime.fromtimestamp(record.created).isoformat(),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
"function": record.funcName,
"line": record.lineno,
}
if hasattr(record, 'extra_data'):
log_record.update(record.extra_data)
return json.dumps(log_record)

logger = logging.getLogger("json_logger")
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Использование с дополнительными данными
extra = {'extra_data': {'user_id': 12345, 'transaction_id': 'abc-123'}}
logger.info("Пользователь совершил платёж", extra=extra)

import logging
import logging.config
import yaml

# Загрузка конфигурации из YAML-файла
with open('logging_config.yaml', 'r') as f:
config = yaml.safe_load(f.read())
logging.config.dictConfig(config)

logger = logging.getLogger(__name__)
logger.debug("Конфигурация загружена из файла")

version: 1
formatters:
standard:
format: '%(asctime)s – %(name)s – %(levelname)s – %(message)s'
error:
format: '%(asctime)s – %(name)s – %(levelname)s – %(message)s – %(pathname)s:%(lineno)d'
handlers:
console:
class: logging.StreamHandler
level: INFO
formatter: standard
file:
class: logging.handlers.RotatingFileHandler
level: ERROR
formatter: error
filename: app_error.log
maxBytes: 10485760
backupCount: 5
loggers:
'': # root logger
level: INFO
handlers: [console, file]
propagate: no
myapp:
level: DEBUG
handlers: [console, file]
propagate: no

import logging

class ContextFilter(logging.Filter):
"""Добавляет информацию о пользователе в лог-записи"""

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

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

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s – User ID: %(user_id)s – %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Добавление фильтра для текущего пользователя
user_filter = ContextFilter(12345)
logger.addFilter(user_filter)

logger.info("Пользователь выполнил операцию")

import logging
import uuid
import contextvars

# Создаем контекстную переменную для хранения trace_id
trace_id_var = contextvars.ContextVar('trace_id', default=None)

class TraceIDFilter(logging.Filter):
def filter(self, record):
trace_id = trace_id_var.get()
if trace_id is None:
trace_id = str(uuid.uuid4())
trace_id_var.set(trace_id)
record.trace_id = trace_id
return True

def configure_logger():
logger = logging.getLogger('app')
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s – [%(trace_id)s] – %(levelname)s – %(message)s')
handler.setFormatter(formatter)

trace_filter = TraceIDFilter()
handler.addFilter(trace_filter)

logger.addHandler(handler)
return logger

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

def process_request(data):
logger.info(f"Получен запрос: {data}")
# Обработка...
logger.info("Запрос обработан")

# Имитация нового запроса – установка нового trace_id
trace_id_var.set(str(uuid.uuid4()))
process_request({"user": "user1", "action": "login"})

import logging
import threading
import queue
import time

class AsyncHandler(logging.Handler):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.log_queue = queue.Queue(-1) # Неограниченная очередь
self._stop_event = threading.Event()
self._thread = threading.Thread(target=self._process_logs)
self._thread.daemon = True
self._thread.start()

def emit(self, record):
try:
self.log_queue.put_nowait(record)
except queue.Full:
pass # В реальном приложении здесь следует обработать ошибку

def _process_logs(self):
actual_handler = logging.FileHandler('async_app.log')
formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
actual_handler.setFormatter(formatter)

while not self._stop_event.is_set():
try:
record = self.log_queue.get(timeout=0.2)
actual_handler.emit(record)
self.log_queue.task_done()
except queue.Empty:
pass
except Exception:
import traceback
traceback.print_exc()

def close(self):
self._stop_event.set()
self._thread.join(timeout=2.0)
super().close()

# Настройка логгера с асинхронным обработчиком
logger = logging.getLogger('async_logger')
logger.setLevel(logging.INFO)
async_handler = AsyncHandler()
logger.addHandler(async_handler)

# Тестирование
for i in range(1000):
logger.info(f"Тестовое сообщение {i}")

print("Все сообщения отправлены в очередь")
time.sleep(2) # Даём время на обработку очереди
async_handler.close()
print("Обработчик закрыт")

import logging
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration

# Настройка интеграции Sentry с логированием
sentry_logging = LoggingIntegration(
level=logging.INFO, # Минимальный уровень для отправки в Sentry
event_level=logging.ERROR # Минимальный уровень для создания событий в Sentry
)

# Инициализация Sentry SDK
sentry_sdk.init(
dsn="https://your_sentry_dsn@sentry.io/project_id",
integrations=[sentry_logging],
traces_sample_rate=1.0, # Трассировка всех запросов
)

# Настройка обычного логгера
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Использование
try:
1 / 0
except Exception as e:
logger.error("Произошла ошибка деления на ноль", exc_info=True)
# Это сообщение будет отправлено и в консоль, и в Sentry

import logging
import json
from datetime import datetime
import socket
from pythonjsonlogger import jsonlogger

class CustomJsonFormatter(jsonlogger.JsonFormatter):
def add_fields(self, log_record, record, message_dict):
super(CustomJsonFormatter, self).add_fields(log_record, record, message_dict)
log_record['timestamp'] = datetime.utcnow().isoformat()
log_record['host'] = socket.gethostname()
log_record['app_name'] = 'my_python_app'

if not log_record.get('severity'):
log_record['severity'] = record.levelname

# Добавление контекстных данных, если они есть
if hasattr(record, 'request_id'):
log_record['request_id'] = record.request_id

# Настройка логгера для отправки в Logstash через файл
logger = logging.getLogger('elk_logger')
logger.setLevel(logging.INFO)

handler = logging.FileHandler('/var/log/my_app/application.log')
formatter = CustomJsonFormatter('%(timestamp)s %(severity)s %(message)s %(host)s %(app_name)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Использование с дополнительной информацией
logger.info("API запрос выполнен успешно", 
extra={"request_id": "req-123", "execution_time": 0.534, "endpoint": "/api/users"})

import logging
import time
from prometheus_client import Counter, Histogram, start_http_server

# Настройка метрик Prometheus
REQUEST_COUNT = Counter('app_request_count', 'Total app requests', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('app_request_latency_seconds', 'Request latency', ['method', 'endpoint'])

# Настройка логгера
logger = logging.getLogger('metrics_logger')
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s – %(levelname)s – %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

# Запуск HTTP-сервера для Prometheus
start_http_server(8000)

def process_request(method, endpoint, status_code):
start_time = time.time()

# Логирование начала запроса
logger.info(f"Получен {method} запрос к {endpoint}")

# Имитация обработки
time.sleep(0.5)

# Измерение времени и обновление метрик
latency = time.time() – start_time
REQUEST_LATENCY.labels(method=method, endpoint=endpoint).observe(latency)
REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status_code).inc()

# Логирование завершения запроса
logger.info(f"Запрос {method} {endpoint} обработан за {latency:.3f}s со статусом {status_code}")

# Имитация запросов
process_request('GET', '/api/users', 200)
process_request('POST', '/api/orders', 201)
process_request('GET', '/api/products', 200)
process_request('GET', '/api/products', 500)

import logging
import structlog

# Настройка structlog
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)

# Создание логгера
logger = structlog.get_logger()

# Использование с контекстными данными
log = logger.bind(user_id=123)
log.info("Пользователь вошел в систему", ip="192.168.1.1", session_id="abc-123")
log.warning("Подозрительная активность", attempts=5, ip="192.168.1.1")
log.error("Ошибка доступа к базе данных", query_time=1.23, database="users_db")