分类 Python 下的文章

记那些踩过的坑:只能用一次的contextlib.contextmanager
最近想要实现一个需求,优雅地使用不阻塞的threading.Lock(),何谓优雅,就是基本不改动,继续使用with。

一开始我的代码是这样子的

def non_blocking_lock(lock: Union[threading.Lock, threading.RLock]):
    @contextmanager
    def _():
        locked = lock.acquire(blocking=False)
        try:
            yield locked
        finally:
            if locked:
                lock.release()

    return _()

测试通过,然而实际使用时却报错: AttributeError: args
翻了翻contextlib的源码,发现在contextlib.py:110,args属性被删除

def __enter__(self):
    # do not keep args and kwds alive unnecessarily
    # they are only needed for recreation, which is not possible anymore
    del self.args, self.kwds, self.func
    try:
        return next(self.gen)
    except StopIteration:
        raise RuntimeError("generator didn't yield") from None

当第二次使用该contextmanager,args已经是不存在的,所以才会抛出AttributeError
所以最后的解决方案很显然:

  1. copy一份contextmanager代码,去掉del那行代码
  2. 手写contextmanager