275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""
|
||
Декораторы для автоматического логирования функций
|
||
"""
|
||
import asyncio
|
||
import inspect
|
||
from functools import wraps
|
||
from typing import Callable, Any, Optional, Dict, Union
|
||
from aiogram.types import Message, CallbackQuery
|
||
|
||
from services.infrastructure.logger import get_logger
|
||
|
||
|
||
def log_function_call(
|
||
function_name: Optional[str] = None,
|
||
log_params: bool = True,
|
||
log_result: bool = False,
|
||
log_level: str = "info",
|
||
quiet: bool = False
|
||
):
|
||
"""
|
||
Декоратор для автоматического логирования входа/выхода из функций
|
||
|
||
Args:
|
||
function_name: Кастомное имя функции для логов (по умолчанию берется из func.__name__)
|
||
log_params: Логировать ли параметры вызова
|
||
log_result: Логировать ли результат выполнения
|
||
log_level: Уровень логирования ('info', 'debug', 'warning')
|
||
quiet: Тихое логирование (только ошибки)
|
||
"""
|
||
def decorator(func: Callable) -> Callable:
|
||
@wraps(func)
|
||
async def async_wrapper(*args, **kwargs):
|
||
logger = get_logger(func.__module__)
|
||
name = function_name or func.__name__
|
||
|
||
# Формируем контекстную информацию
|
||
context_info = _build_context_info(args, kwargs, log_params)
|
||
|
||
# Логируем вход в функцию (только если не тихий режим)
|
||
if not quiet:
|
||
log_method = getattr(logger, log_level)
|
||
log_method(f"🚀 Начало выполнения {name}{context_info}")
|
||
|
||
try:
|
||
result = await func(*args, **kwargs)
|
||
|
||
# Логируем успешное завершение (только если не тихий режим)
|
||
if not quiet:
|
||
result_info = ""
|
||
if log_result and result is not None:
|
||
result_info = f" | Результат: {_format_result(result)}"
|
||
|
||
log_method(f"✅ Успешное завершение {name}{result_info}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
# Логируем ошибку (всегда, даже в тихом режиме)
|
||
logger.error(f"❌ Ошибка в {name}: {e}")
|
||
raise
|
||
|
||
@wraps(func)
|
||
def sync_wrapper(*args, **kwargs):
|
||
logger = get_logger(func.__module__)
|
||
name = function_name or func.__name__
|
||
|
||
# Формируем контекстную информацию
|
||
context_info = _build_context_info(args, kwargs, log_params)
|
||
|
||
# Логируем вход в функцию (только если не тихий режим)
|
||
if not quiet:
|
||
log_method = getattr(logger, log_level)
|
||
log_method(f"🚀 Начало выполнения {name}{context_info}")
|
||
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
|
||
# Логируем успешное завершение (только если не тихий режим)
|
||
if not quiet:
|
||
result_info = ""
|
||
if log_result and result is not None:
|
||
result_info = f" | Результат: {_format_result(result)}"
|
||
|
||
log_method(f"✅ Успешное завершение {name}{result_info}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
# Логируем ошибку (всегда, даже в тихом режиме)
|
||
logger.error(f"❌ Ошибка в {name}: {e}")
|
||
raise
|
||
|
||
# Возвращаем правильный wrapper в зависимости от типа функции
|
||
return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
|
||
return decorator
|
||
|
||
|
||
def log_business_event(
|
||
event_name: str,
|
||
log_params: bool = True,
|
||
log_result: bool = True
|
||
):
|
||
"""
|
||
Декоратор для логирования бизнес-событий
|
||
|
||
Args:
|
||
event_name: Название бизнес-события
|
||
log_params: Логировать ли параметры
|
||
log_result: Логировать ли результат
|
||
"""
|
||
def decorator(func: Callable) -> Callable:
|
||
@wraps(func)
|
||
async def async_wrapper(*args, **kwargs):
|
||
logger = get_logger(func.__module__)
|
||
|
||
# Формируем контекстную информацию
|
||
context_info = _build_context_info(args, kwargs, log_params)
|
||
|
||
# Логируем бизнес-событие
|
||
logger.info(f"📊 Бизнес-событие: {event_name}{context_info}")
|
||
|
||
try:
|
||
result = await func(*args, **kwargs)
|
||
|
||
# Логируем результат бизнес-события
|
||
if log_result and result is not None:
|
||
result_info = _format_result(result)
|
||
logger.info(f"📈 Результат {event_name}: {result_info}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.error(f"💥 Ошибка в бизнес-событии {event_name}: {e}")
|
||
raise
|
||
|
||
@wraps(func)
|
||
def sync_wrapper(*args, **kwargs):
|
||
logger = get_logger(func.__module__)
|
||
|
||
# Формируем контекстную информацию
|
||
context_info = _build_context_info(args, kwargs, log_params)
|
||
|
||
# Логируем бизнес-событие
|
||
logger.info(f"📊 Бизнес-событие: {event_name}{context_info}")
|
||
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
|
||
# Логируем результат бизнес-события
|
||
if log_result and result is not None:
|
||
result_info = _format_result(result)
|
||
logger.info(f"📈 Результат {event_name}: {result_info}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.error(f"💥 Ошибка в бизнес-событии {event_name}: {e}")
|
||
raise
|
||
|
||
return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
|
||
return decorator
|
||
|
||
|
||
def log_fsm_transition(
|
||
from_state: Optional[str] = None,
|
||
to_state: Optional[str] = None
|
||
):
|
||
"""
|
||
Декоратор для логирования переходов FSM состояний
|
||
|
||
Args:
|
||
from_state: Исходное состояние (если None, будет определено автоматически)
|
||
to_state: Целевое состояние (если None, будет определено автоматически)
|
||
"""
|
||
def decorator(func: Callable) -> Callable:
|
||
@wraps(func)
|
||
async def async_wrapper(*args, **kwargs):
|
||
logger = get_logger(func.__module__)
|
||
|
||
# Извлекаем FSM context из аргументов
|
||
fsm_context = None
|
||
for arg in args:
|
||
if hasattr(arg, 'get_state') and hasattr(arg, 'set_state'):
|
||
fsm_context = arg
|
||
break
|
||
|
||
# Логируем переход состояния
|
||
if fsm_context:
|
||
current_state = await fsm_context.get_state()
|
||
logger.info(f"🔄 FSM переход: {current_state} -> {to_state or 'новое состояние'}")
|
||
|
||
try:
|
||
result = await func(*args, **kwargs)
|
||
|
||
# Логируем успешный переход
|
||
if fsm_context:
|
||
new_state = await fsm_context.get_state()
|
||
logger.info(f"✅ FSM переход завершен: {from_state or 'предыдущее состояние'} -> {new_state}")
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка в FSM переходе: {e}")
|
||
raise
|
||
|
||
return async_wrapper
|
||
return decorator
|
||
|
||
|
||
def _build_context_info(args: tuple, kwargs: dict, log_params: bool) -> str:
|
||
"""Построение контекстной информации для логов"""
|
||
if not log_params:
|
||
return ""
|
||
|
||
context_parts = []
|
||
|
||
# Извлекаем информацию о пользователе из аргументов
|
||
user_id = None
|
||
for arg in args:
|
||
if isinstance(arg, (Message, CallbackQuery)):
|
||
user_id = arg.from_user.id
|
||
context_parts.append(f"user_id={user_id}")
|
||
break
|
||
elif hasattr(arg, 'from_user_id'):
|
||
user_id = arg.from_user_id
|
||
context_parts.append(f"from_user_id={user_id}")
|
||
elif hasattr(arg, 'to_user_id'):
|
||
context_parts.append(f"to_user_id={arg.to_user_id}")
|
||
elif hasattr(arg, 'id') and isinstance(arg.id, int):
|
||
context_parts.append(f"id={arg.id}")
|
||
|
||
# Добавляем важные параметры из kwargs
|
||
important_params = ['question_id', 'user_id', 'page', 'limit', 'status']
|
||
for param in important_params:
|
||
if param in kwargs and kwargs[param] is not None:
|
||
context_parts.append(f"{param}={kwargs[param]}")
|
||
|
||
return f" | {', '.join(context_parts)}" if context_parts else ""
|
||
|
||
|
||
def _format_result(result: Any) -> str:
|
||
"""Форматирование результата для логов"""
|
||
if result is None:
|
||
return "None"
|
||
|
||
if isinstance(result, (str, int, float, bool)):
|
||
return str(result)
|
||
|
||
if hasattr(result, 'id'):
|
||
return f"id={result.id}"
|
||
|
||
if isinstance(result, (list, tuple)):
|
||
return f"count={len(result)}"
|
||
|
||
if isinstance(result, dict):
|
||
return f"keys={list(result.keys())}"
|
||
|
||
return str(type(result).__name__)
|
||
|
||
|
||
# Удобные алиасы для часто используемых декораторов
|
||
log_handler = log_function_call
|
||
log_service = log_function_call
|
||
log_business = log_business_event
|
||
log_fsm = log_fsm_transition
|
||
|
||
# Тихие декораторы для middleware и служебных функций
|
||
log_quiet = lambda **kwargs: log_function_call(quiet=True, **kwargs)
|
||
log_middleware = lambda **kwargs: log_function_call(quiet=True, log_level="debug", **kwargs)
|
||
|
||
# Декоратор для служебных функций (только ошибки)
|
||
def log_utility(func: Callable) -> Callable:
|
||
"""Декоратор для служебных функций - логирует только ошибки"""
|
||
return log_function_call(quiet=True, log_params=False, log_result=False)(func)
|