记那些踩过的坑:只能用一次的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
所以最后的解决方案很显然:
- copy一份contextmanager代码,去掉del那行代码
- 手写contextmanager
本文由 hunsh 创作,采用 知识共享署名4.0 国际许可协议进行许可
本站文章除注明转载/出处外,均为本站原创或翻译,转载前请务必署名
最后编辑时间为: Aug 26, 2019 at 09:12 pm