Using python multiprocessing.Lock as an argument type in mypy

Question

I have a python function that takes in a multiprocessing.Lock object as an input and uses the acquire and release functions on it. Evaluating it with mypy returns the error Function multiprocessing.Lock" is not valid as a type. How can I properly annotate this function?

Answer 1

Lock is actually a plain function that returns a _LockType, which is defined as _LockType = synchronize.Lock, so you can do:

from multiprocessing.synchronize import Lock as LockBase

def func(lock: LockBase):
    pass

Although, the fact that they have _LockType as module private suggests they're treating it as an implementation detail that may change in the future, so be careful.

---

There's also this comment block above where _LockType is defined:

# 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

For multiprocessing factory functions Lock() and RLock(), there are at least corresponding classes in the multiprocessing.synchronize submodule.

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

The situation is worse in the closely related threading package, which might need to solve this internally, but till that's done, one can get away with defining a little protocol for the two locks:

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

This protocol would be universally suitable for the locks from both the threading and multiprocessing packages, but it's not, because the multiprocessing.Lock.acquire has block as the 1st argument, while threading.Lock.acquire has blocking. I wonder if that's intentional.