【问题标题】:How should I catch exceptions in a decorator function that can raise, if the decorator function is in a library I can't modify?如果装饰器函数在我无法修改的库中,我应该如何在可以引发的装饰器函数中捕获异常?
【发布时间】:2015-05-07 19:07:34
【问题描述】:

我正在使用 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__ 方法允许将其用作装饰器,如下所示:

从 statsd 导入 StatsClient statsd = StatsClient() @statsd.timer('myfunc') def myfunc(a, b): """计算 a 和 b 能做的最复杂的事情。"""

我没有简单的能力来修改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) 类 Wat(对象): def __call__(self, *args, **kwargs): timer_instance = stats_client.timer(*args, **kwargs) 尝试: 返回 timer_instance.__call__(wrapped_func)(*args, **kwargs) 除了例外: logger.warning("捕获异常", exc_info=True) 定义 foo(): 经过 返回 foo 返回笏()

然后会像这样使用:

@my_timer_wrapper('stastd_timer_name') def timed_func(): 做工作()

有没有更好或更 Pythonic 的方式?

【问题讨论】:

  • 看起来该类可以用作上下文管理器,这样更容易包装在 try/except 中。
  • 你可以猴子补丁statsd。在 appengine_config.py 中执行修补,因此只需要在您的代码库中执行一次。

标签: python sockets google-app-engine decorator statsd


【解决方案1】:

看起来这是一个“尽可能简单”的案例 新的装饰器在你的计时器装饰器周围添加了一个额外的尝试/除外。

唯一的问题是接受参数的装饰器需要 要定义的 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。

【讨论】:

    猜你喜欢
    • 2015-09-13
    • 2013-08-22
    • 2019-10-05
    • 1970-01-01
    • 2018-10-25
    • 2011-12-01
    • 1970-01-01
    • 2016-12-14
    • 2020-05-20
    相关资源
    最近更新 更多