使用 python multiprocessing.Lock 作为 mypy 中的参数类型

Question

我有一个 python 函数，它接受 multiprocessing.Lock 对象作为输入，并在其上使用 acquire 和release 函数。使用 mypy 评估它会返回错误

Function multiprocessing.Lock" is not valid as a type

。我怎样才能正确注释这个函数？

Answer 1

Lock

实际上是一个普通函数，返回一个

_LockType

，它被定义为

_LockType = synchronize.Lock

，所以你可以这样做：

from multiprocessing.synchronize import Lock as LockBase

def func(lock: LockBase):
    pass

尽管如此，他们将

_LockType

作为模块私有这一事实表明他们将其视为将来可能会发生变化的实现细节，所以要小心。

---

上面还有这个注释块，其中定义了

_LockType

：

# The following type aliases can be used to annotate the return values of
# the corresponding functions. They are not defined at runtime.
#
# from multiprocessing import Lock
# from typing import TYPE_CHECKING
# if TYPE_CHECKING:
#     from multiprocessing import _LockType
# lock: _LockType = Lock() 
. . .
_LockType = synchronize.Lock

Answer 2

对于

multiprocessing

工厂函数

Lock()

和

RLock()

，

multiprocessing.synchronize

子模块中至少有相应的类。

from multiprocessing.synchronize import Lock as LockT, RLock as RLockT

密切相关的

threading

包中的情况更糟，可能需要在内部解决这个问题，但在完成之前，可以为两个锁定义一些 protocol ：

class _LockType(contextlib.AbstractContextManager, typing.Protocol):
    def acquire(self, blocking: bool = ..., timeout: float = ...) -> bool: ...
    def release(self) -> None: ...


def ensure_lock(lock: _LockType | None) -> _LockType:
    return threading.Lock() if lock is None else lock

lock1 = ensure_lock(None)
lock2 = ensure_lock(Lock())
rlock = ensure_lock(RLock())
lock1.acquire()
lock1.release()  # mypy (1.7.1) is happy about all of the above

该协议普遍适用于

threading

和

multiprocessing

包中的锁，但事实并非如此，因为

multiprocessing.Lock.acquire

将

block

作为第一个参数，而

threading.Lock.acquire

则将

 blocking

。我想知道这是不是故意的。