首页 文章

我应该如何捕获可以引发的装饰器函数中的异常,如果装饰器函数在库中我无法修改?

提问于
浏览
2

我在Google App Engine(GAE)上工作python statsd library . 不幸的是,GAE可以在使用套接字时不时提升 ApplicationError: 4 Unknown error. . 错误是 apiproxy_errors.ApplicationError .

statsd客户端已经设置为捕获 socket.error ,但不是套接字可以在GAE上引发的 ApplicationError .

我特意使用 timer ,它返回 Timer 的实例:https://github.com/jsocol/pystatsd/blob/master/statsd/client.py#L13

Timer__call__ 方法允许它用作装饰器,如下所示:

from statsd import StatsClient

statsd = StatsClient()

@statsd.timer('myfunc')
def myfunc(a, b):
    """Calculate the most complicated thing a and b can do."""

我没有简单的能力来修改 Timer.__call__ 方法本身以简单地捕获 ApplicationError .

我应该如何编写一个包装器或其他装饰器仍然允许像 @my_timer_wrapper('statsd_timer_name') 这样的干净装饰,但是它会捕获包装/装饰 timer 方法中可能出现的其他异常?

这是我的代码库中的基础模块,将在许多地方使用(无论我们想要什么时间) . 因此虽然this SO answer可能有用,但我真的想避免在我的代码库中强制使用 @statsclient.timer 本身在 try-except 块中定义 .

我正在考虑做以下事情:

def my_timer_wrapper(wrapped_func, *args, **kwargs):
  @functools.wraps(wrapped_func)
  class Wat(object):
    def __call__(self, *args, **kwargs):
      timer_instance = stats_client.timer(*args, **kwargs)
      try:
        return timer_instance.__call__(wrapped_func)(*args, **kwargs)
      except Exception:
        logger.warning("Caught exception", exc_info=True)
        def foo():
          pass
        return foo

  return Wat()

然后将使用如下:

@my_timer_wrapper('stastd_timer_name')
def timed_func():
  do_work()

是否有更好或更多的pythonic方式?

1 回答

  • 0

    它看起来像是一个“尽可能简单”的新装饰器,在你的计时器装饰器周围添加一个额外的尝试/除外 .

    唯一的问题是装饰器需要定义需要2级嵌套函数的参数,几乎总是使它们看起来很复杂,即使它们不是:

    from functools import wraps
    
    def  shielded_timer(statsd, 
                        exceptions=(apiproxy_errors.ApplicationError),
                        name=None):
    
        def decorator(func):
            timer_decorator = statsd.timer(name or func.__name__)
            new_func = timer_decorator(func)
            @wraps(func)
            def wrapper(*args, **kw):
                try:
                    return new_func(*args, **kw)
                except BaseException as error:
                    if isinstance (error, exceptions):
                        # Expected error (ApplicationError by default) ocurred
                        pass
                    else:
                        raise
            return wrapper
        return decorator
    
    
    
    #####################
    statsd = StatsClient()
    @shielded_timer(statsd)
    def my_func(a,b):
        ...
    

    正如你所看到的那样,即使包含额外的细节也很容易 - 在这种情况下,我已经在装饰时可以配置所需的异常,并且可选地,它在调用statsd.timer时自动使用装饰函数的名称 .

相关问题