Python3。如何优化装饰器来处理异常?
Python3. How to optimize decorator for handling exceptions?
我有装饰器用作异常处理程序。我想优化它,因为与简单的 try...catch 相比,它慢了大约 6 倍。
我的装饰器代码:
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):
self.func = func
return self.wrapper
def wrapper(self, *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)
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
作为基准,我使用带有 try...catch 的简单函数和带有装饰器的函数:
# simple function
def divide(a, b):
try:
return a // b
except ZeroDivisionError:
return 'error'
# function with decorator
@ProcessException({ZeroDivisionError: lambda: 'err'})
def divide2(a, b):
return a // b
简单函数迭代 10000 次的结果:
timeit.timeit('divide(1, 0)', number=10000, setup='from __main__ import divide')
0.010692662000110431
并使用装饰器函数:
timeit.timeit('divide2(1, 0)', number=10000, setup='from __main__ import divide2')
0.053688491000002614
请帮我优化一下,请说明瓶颈在哪里?
好吧,我终于优化了我的装饰器。所以,代码(下面的解释):
from inspect import iscoroutinefunction
from asyncio import QueueEmpty, QueueFull
from concurrent.futures import TimeoutError
class ProcessException(object):
__slots__ = ('func', 'handlers')
def __init__(self, custom_handlers=None):
self.func = None
if isinstance(custom_handlers, property):
custom_handlers = custom_handlers.__get__(self, self.__class__)
def raise_exception(e: Exception):
raise e
exclude = {
QueueEmpty: lambda e: None,
QueueFull: lambda e: None,
TimeoutError: lambda e: None
}
self.handlers = {
**exclude,
**(custom_handlers or {}),
Exception: raise_exception
}
def __call__(self, func):
self.func = func
if iscoroutinefunction(self.func):
def wrapper(*args, **kwargs):
return self._coroutine_exception_handler(*args, **kwargs)
else:
def wrapper(*args, **kwargs):
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:
return self.handlers.get(e.__class__, Exception)(e)
def _sync_exception_handler(self, *args, **kwargs):
try:
return self.func(*args, **kwargs)
except Exception as e:
return self.handlers.get(e.__class__, Exception)(e)
首先,我使用cProfile 来查看decorator 进行了哪些调用。我注意到 asyncio.iscoroutinefunction 打了额外的电话。我已经删除了它。我还删除了所有额外的 if
并为所有处理程序创建了通用字典。我的解决方案仍然不如 try...catch 快,但现在它运行得更快了。
我有装饰器用作异常处理程序。我想优化它,因为与简单的 try...catch 相比,它慢了大约 6 倍。
我的装饰器代码:
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):
self.func = func
return self.wrapper
def wrapper(self, *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)
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
作为基准,我使用带有 try...catch 的简单函数和带有装饰器的函数:
# simple function
def divide(a, b):
try:
return a // b
except ZeroDivisionError:
return 'error'
# function with decorator
@ProcessException({ZeroDivisionError: lambda: 'err'})
def divide2(a, b):
return a // b
简单函数迭代 10000 次的结果:
timeit.timeit('divide(1, 0)', number=10000, setup='from __main__ import divide')
0.010692662000110431
并使用装饰器函数:
timeit.timeit('divide2(1, 0)', number=10000, setup='from __main__ import divide2')
0.053688491000002614
请帮我优化一下,请说明瓶颈在哪里?
好吧,我终于优化了我的装饰器。所以,代码(下面的解释):
from inspect import iscoroutinefunction
from asyncio import QueueEmpty, QueueFull
from concurrent.futures import TimeoutError
class ProcessException(object):
__slots__ = ('func', 'handlers')
def __init__(self, custom_handlers=None):
self.func = None
if isinstance(custom_handlers, property):
custom_handlers = custom_handlers.__get__(self, self.__class__)
def raise_exception(e: Exception):
raise e
exclude = {
QueueEmpty: lambda e: None,
QueueFull: lambda e: None,
TimeoutError: lambda e: None
}
self.handlers = {
**exclude,
**(custom_handlers or {}),
Exception: raise_exception
}
def __call__(self, func):
self.func = func
if iscoroutinefunction(self.func):
def wrapper(*args, **kwargs):
return self._coroutine_exception_handler(*args, **kwargs)
else:
def wrapper(*args, **kwargs):
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:
return self.handlers.get(e.__class__, Exception)(e)
def _sync_exception_handler(self, *args, **kwargs):
try:
return self.func(*args, **kwargs)
except Exception as e:
return self.handlers.get(e.__class__, Exception)(e)
首先,我使用cProfile 来查看decorator 进行了哪些调用。我注意到 asyncio.iscoroutinefunction 打了额外的电话。我已经删除了它。我还删除了所有额外的 if
并为所有处理程序创建了通用字典。我的解决方案仍然不如 try...catch 快,但现在它运行得更快了。