记那些踩过的坑:只能用一次的contextlib.contextmanager

记那些踩过的坑:只能用一次的 contextlib.contextmanager

最近想要实现一个需求,优雅地使用不阻塞的 threading.Lock(),何谓优雅,就是基本不改动,继续使用 with。

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

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 属性被删除

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

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


记那些踩过的坑:只能用一次的contextlib.contextmanager
https://hunsh.net/archives/3/
发布于
2019年8月26日
许可协议