Python v3.x: обработчик исключений для корутин и синхронных функций. Вобщем, для всего

В свободное время я работаю над своим небольшим проектом. Написан на Python v3.x + SQLAlchemy. Возможно, я когда-нибудь напишу и о нем, но сегодня хочу рассказать о своем декораторе для обработки исключений. Его можно применять как для функций, так и для методов. Синхронных и асинхронных. Также можно подключать кастомные хэндлеры исключений.

Декоратор на текущий момент выглядит так:

import asyncio

from asyncio import QueueEmpty, QueueFull
from concurrent.futures import TimeoutError


class ProcessException(object):

    __slots__ = ('func', 'custom_handlers', 'exclude')

    def __init__(self, custom_handlers=None):
        self.func = None
        self.custom_handlers: dict = custom_handlers
        self.exclude = [QueueEmpty, QueueFull, TimeoutError]

    def __call__(self, func, *a):
        self.func = func

        def wrapper(*args, **kwargs):
            if self.custom_handlers:
                if isinstance(self.custom_handlers, property):
                    self.custom_handlers = self.custom_handlers.__get__(self, self.__class__)

            if asyncio.iscoroutinefunction(self.func):
                return self._coroutine_exception_handler(*args, **kwargs)
            else:
                return self._sync_exception_handler(*args, **kwargs)

        return wrapper

    async def _coroutine_exception_handler(self, *args, **kwargs):
        try:
            return await self.func(*args, **kwargs)
        except Exception as e:
            if self.custom_handlers and e.__class__ in self.custom_handlers:
                return self.custom_handlers[e.__class__]()

            if e.__class__ not in self.exclude:
                raise e

    def _sync_exception_handler(self, *args, **kwargs):
        try:
            return self.func(*args, **kwargs)
        except Exception as e:
            if self.custom_handlers and e.__class__ in self.custom_handlers:
                return self.custom_handlers[e.__class__]()

            if e.__class__ not in self.exclude:
                raise e

Разберем по порядку. __slots__ я использую для небольшой экономии памяти. Бывает полезно, если объект используется ну ооочень часто.

На этапе инициализации в __init__ мы сохраняем custom_handlers (в случае, если понадобилось их передать). На всякий случай обозначил, что мы ожидаем там увидеть словарь, хотя, возможно, в будущем, есть смысл добавить пару жестких проверок. В свойстве self.exclude лежит список исключений, которые обрабатывать не нужно. В случае такого исключения функция с декоратором вернет None. На текущий момент список заточен под мой проект и возможно есть смысл его вынести в отдельный конфиг.

Самое главное происходит в __call__. Поэтому при использовании декоратора его нужно вызывать. Даже без параметров:

@ProcessException()
def some_function(*args):
    return None

Т.е. вот так уже будет неправильно и будет вызвана ошибка:

@ProcessException
def some_function(*args):
    return None

В этом случае мы получим текущую функцию, которую, в зависимости от степени ее асинхронности обработаем либо как обычную синхронную, либо как корутин.

На что здесь можно обратить внимание. Первое, это проверка на проперти:

if self.custom_handlers:
    if isinstance(self.custom_handlers, property):
        self.custom_handlers = self.custom_handlers.__get__(self, self.__class__)

Зачем я это делаю.

Конечно же 
         не потому, что 
                      я айти-Маяковский 
                                   и мне платят построчно.

Два if здесь для улучшения читабельности (да-да, ведь код может саппортить человек с садистскими наклонностями), а self.custom_handlers.__get__(self, self.__class__) мы делаем для того, чтобы не терять класс, в случае, если мы решили хэндлеры хранить в @property класса.

Например, так:

class Math(object):
    @property
    def exception_handlers(self):
        return {
            ZeroDivisionError: lambda: 'Делить на ноль нельзя, но можно умножить'
        }
    
    @ProcessException(exception_handlers)
    def divide(self, a, b):
        return a // b

Если не сделать self.custom_handlers.__get__(...), то вместо содержимого @property мы будем получать что-то типа <property object at 0x7f78d844f9b0>.

Собственно, в примере выше показан способ подключения кастомных хэндлеров. В общем случае это делается так:

@ProcessException({ZeroDivisionError: lambda: 'Делить на ноль можно, но с ошибкой'})
def divide(a, b):
    return a // b

В случае с классом (если мы собираемся передавать свойства/методы) нужно учесть, что на этапе инициализации декоратора класса как такового еще нету и методы/свойства суть простые функции. Поэтому мы можем передать только то, что объявлено выше. Поэтому вариант с @property — это возможность применять через self все функции, которые ниже по коду. Ну либо можно использовать лямбды, если self не нужен.

Для асинхронного кода справедливы все вышеописанные примеры.

Напоследок хочу обратить внимание, что если исключение на своем пути не встретило кастомных хэндлеров, то оно просто рейзится (raise) дальше.

Жду ваших комментариев. Спасибо за то, что уделили внимание моей статье.