首页 > 解决方案 > 从函数中收集警告的最 Pythonic 方式

问题描述

考虑一个非常简单的函数:

def generate_something(data):
    if data is None:
        raise Exception('No data!')

    return MyObject(data)

它的输出基本上是我想要创建的对象的一个​​实例,或者如果函数无法创建该对象,则会出现异常。我们可以说输出是二进制的,因为它要么成功(并返回一个对象),要么不成功(并返回一个异常)。

处理第三种状态的最 Pythonic 方式是什么,即“成功但有一些警告”?

def generate_something(data):
    warnings = []

    if data is None:
        raise Exception("No data!")

    if data.value_1 == 2:
        warnings.append('Hmm, value_1 is 2')    

    if data.value_2 == 1:
        warnings.append('Hmm, value_2 is 1')    

    return MyObject(data), warnings

返回一个元组是处理这个问题的唯一方法,还是可以从函数内广播或产生警告并从调用者那里捕获它们?

标签: pythonpython-3.x

解决方案


内置选项:warnings

warningsPython 在模块中实现了内置的警告机制。这样做的问题是warnings维护一个全局警告过滤器,这可能会无意中导致您的函数抛出的警告被抑制。这是问题的演示:

import warnings

def my_func():
    warnings.warn('warning!')

my_func()  # prints "warning!"

warnings.simplefilter("ignore")
my_func()  # prints nothing

如果你想warnings不管这个,你可以使用warnings.catch_warnings(record=True)收集所有抛出的警告在一个列表中:

with warnings.catch_warnings(record=True) as warning_list:
    warnings.warn('warning 3')

print(warning_list)  # output: [<warnings.WarningMessage object at 0x7fd5f2f484e0>]

自制的选择

出于上述原因,我建议改为使用您自己的警告机制。有多种方法可以实现这一点:

  • 只需返回警告列表

    开销最少的最简单解决方案:只需返回警告。

    def example_func():
        warnings = []
    
        if ...:
            warnings.append('warning!')
    
        return result, warnings
    
    result, warnings = example_func()
    for warning in warnings:
        ...  # handle warnings
    
  • 将警告处理程序传递给函数

    如果您想在生成警告时立即处理它们,您可以重写您的函数以接受警告处理程序作为参数:

    def example_func(warning_handler=lambda w: None):
        if ...:
            warning_handler('warning!')
    
        return result
    
    
    def my_handler(w):
        print('warning', repr(w), 'was produced')
    
    result = example_func(my_handler)
    
  • contextvars(蟒蛇3.7+)

    在 python 3.7 中,我们得到了这个contextvars模块,它让我们可以基于上下文管理器实现更高级别的警告机制:

    import contextlib
    import contextvars
    import warnings
    
    
    def default_handler(warning):
        warnings.warn(warning, stacklevel=3)
    
    _warning_handler = contextvars.ContextVar('warning_handler', default=default_handler)
    
    
    def warn(msg):
        _warning_handler.get()(msg)
    
    
    @contextlib.contextmanager
    def warning_handler(handler):
        token = _warning_handler.set(handler)
        yield
        _warning_handler.reset(token)
    

    使用示例:

    def my_warning_handler(w):
        print('warning', repr(w), 'was produced')
    
    with warning_handler(my_warning_handler):
        warn('some problem idk')  # prints "warning 'some problem idk' was produced"
    
    warn(Warning('another problem'))  # prints "Warning: another problem"
    

    警告:截至目前,contextvars不支持生成器。(相关PEP。)类似以下示例的内容将无法正常工作:

    def gen(x):
        with warning_handler(x):
            for _ in range(2):
                warn('warning!')
                yield
    
    g1 = gen(lambda w: print('handler 1'))
    g2 = gen(lambda w: print('handler 2'))
    
    next(g1)  # prints "handler 1"
    next(g2)  # prints "handler 2"
    next(g1)  # prints "handler 2"
    
  • 没有contextvars(对于python <3.7)

    如果你没有contextvars,你可以使用这个 async-unsafe 实现来代替:

    import contextlib
    import threading
    import warnings
    
    
    def default_handler(warning):
        warnings.warn(warning, stacklevel=3)
    
    _local_storage = threading.local()
    _local_storage.warning_handler = default_handler
    
    
    def _get_handler():
        try:
            return _local_storage.warning_handler
        except AttributeError:
            return default_handler
    
    
    def warn(msg):
        handler = _get_handler()
        handler(msg)
    
    
    @contextlib.contextmanager
    def warning_handler(handler):
        previous_handler = _get_handler()
        _local_storage.warning_handler = handler
    
        yield
    
        _local_storage.warning_handler = previous_handler
    

推荐阅读