src.core.err

  1"""Функционал обработки ошибок"""
  2
  3import logging
  4import os
  5from collections import deque
  6from functools import wraps
  7
  8from aiogram.exceptions import AiogramError, TelegramNetworkError
  9from redis import exceptions as rexc
 10from redis.asyncio.client import Pipeline
 11
 12from core.exceptions import DatabaseError
 13
 14ERRORS = deque([], maxlen=5)
 15
 16logger = logging.getLogger()
 17rLogger = logging.getLogger("redis")
 18
 19
 20def log_cash_error(error: Exception):
 21    error_info = {"type": type(error).__name__, "message": str(error)}
 22
 23    if error_info not in ERRORS:
 24        ERRORS.append(error_info)
 25
 26        return True
 27
 28
 29def get_args_dict(fn, args, kwargs):
 30    """Создает словарь аргументов для функции.
 31
 32    Args:
 33        fn (function): Функция, для которой необходимо получить аргументы.
 34        args (tuple): Позиционные аргументы функции.
 35        kwargs (dict): Именованные аргументы функции.
 36
 37    Returns:
 38        dict: Словарь аргументов, где ключи - имена параметров, а значения - переданные значения.
 39    """
 40    args_names = fn.__code__.co_varnames[: fn.__code__.co_argcount]
 41    return {**dict(zip(args_names, args)), **kwargs}
 42
 43
 44def exception_logging(
 45    ignore_raise=False,
 46    custom_exception=Exception,
 47    message="Something went wrong",
 48):
 49    """Декоратор для логирования исключений, возникающих в синхронных функциях.
 50
 51    Args:
 52        ignore_raise (bool, optional): Прерывать ли выполнение программы при возникновении исключения. Defaults to False.
 53        custom_exception (Exception, optional): Тип исключения, который нужно обрабатывать. Defaults to Exception.
 54        message (str, optional): Сообщение для логирования при возникновении исключения. Defaults to "Something went wrong".
 55
 56    Returns:
 57        function: Обернутая функция с логированием исключений.
 58    """
 59
 60    def __exception_logging(func):
 61        old_factory = logging.getLogRecordFactory()
 62
 63        def record_factory(*args, **kwargs):
 64            """Создает запись лога с информацией о функции."""
 65            record = old_factory(*args, **kwargs)
 66            if hasattr(func, "__wrapped__"):
 67                record.funcName = func.__wrapped__.__name__
 68                record.filename = os.path.basename(
 69                    func.__wrapped__.__code__.co_filename
 70                )
 71                record.lineno = func.__wrapped__.__code__.co_firstlineno
 72            else:
 73                record.funcName = func.__name__
 74                record.filename = os.path.basename(func.__code__.co_filename)
 75                record.lineno = func.__code__.co_firstlineno
 76            return record
 77
 78        @wraps(func)
 79        def wrapper(*args, **kwargs):
 80            """Обертка для функции с обработкой исключений."""
 81            try:
 82                result = func(*args, **kwargs)
 83            except custom_exception as e:
 84                logging.setLogRecordFactory(record_factory)
 85                signature = get_args_dict(func, args, kwargs)
 86                logger.debug(f"function {func.__name__} called with args {signature}")
 87                logger.exception(f"{message}\nError: {e}")
 88                logging.setLogRecordFactory(old_factory)
 89                if not ignore_raise:
 90                    raise
 91            else:
 92                return result
 93
 94        return wrapper
 95
 96    return __exception_logging
 97
 98
 99def redis_exceptor(func):
100    """Декоратор для обработки исключений при работе с Redis.
101
102    Args:
103        func (function): Функция, которая будет обернута для обработки исключений.
104
105    Returns:
106        function: Обернутая асинхронная функция с обработкой исключений Redis.
107    """
108
109    @wraps(func)
110    async def wrapper(pipeline: Pipeline):
111        """Обертка для функции с обработкой исключений Redis."""
112        try:
113            logQuery = str(getattr(pipeline, "command_stack", ""))
114
115            result = await func(pipeline)
116            return result
117
118        except rexc.AuthenticationError:
119            rLogger.exception(
120                f"Ошибка аутентификации при подключении к Redis :: {logQuery}"
121            )
122            raise DatabaseError
123        except rexc.ConnectionError:
124            rLogger.exception(f"Ошибка подключения к Redis :: {logQuery}")
125            raise DatabaseError
126        except rexc.TimeoutError:
127            rLogger.exception(
128                f"Превышено время ожидания при работе с Redis :: {logQuery}"
129            )
130            raise DatabaseError
131        except IndexError:
132            rLogger.exception("Command Stack is empty")
133            raise DatabaseError
134        except rexc.RedisError:
135            rLogger.exception(f"Произошла неизвестная ошибка c Redis :: {logQuery}")
136            raise DatabaseError
137
138    return wrapper
139
140
141def bot_except(func):
142    """Декоратор для логирования исключений, возникающих в асинхронных функциях."""
143
144    @wraps(func)
145    async def wrapper(*args, **kwargs):
146        try:
147            result = await func(*args, **kwargs)
148
149        except TelegramNetworkError as e:
150            logger.warning(e.args[0])
151        except AiogramError:
152            logger.exception("Ошибка Aiogram API")
153            raise
154        except Exception:
155            logger.exception("Неизвестная ошибка")
156            raise
157        else:
158            return result
159
160    return wrapper