Metaclass that handles singleton resolution when is_singleton=True. Singleton logic lives here rather than in ReadWriteLock.get_lock so that ``ReadWriteLock(path)`` transparently returns cached instances without a 2-arg ``super()`` call that type checkers cannot verify.
| 57 | |
| 58 | |
| 59 | class _ReadWriteLockMeta(type): |
| 60 | """ |
| 61 | Metaclass that handles singleton resolution when is_singleton=True. |
| 62 | |
| 63 | Singleton logic lives here rather than in ReadWriteLock.get_lock so that ``ReadWriteLock(path)`` transparently |
| 64 | returns cached instances without a 2-arg ``super()`` call that type checkers cannot verify. |
| 65 | |
| 66 | """ |
| 67 | |
| 68 | _instances: WeakValueDictionary[pathlib.Path, ReadWriteLock] |
| 69 | _instances_lock: threading.Lock |
| 70 | |
| 71 | def __call__( |
| 72 | cls, |
| 73 | lock_file: str | os.PathLike[str], |
| 74 | timeout: float = -1, |
| 75 | *, |
| 76 | blocking: bool = True, |
| 77 | is_singleton: bool = True, |
| 78 | ) -> ReadWriteLock: |
| 79 | if not is_singleton: |
| 80 | return super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) |
| 81 | |
| 82 | normalized = pathlib.Path(lock_file).resolve() |
| 83 | with cls._instances_lock: |
| 84 | if normalized not in cls._instances: |
| 85 | instance = super().__call__(lock_file, timeout, blocking=blocking, is_singleton=is_singleton) |
| 86 | cls._instances[normalized] = instance |
| 87 | else: |
| 88 | instance = cls._instances[normalized] |
| 89 | |
| 90 | if instance.timeout != timeout or instance.blocking != blocking: |
| 91 | msg = ( |
| 92 | f"Singleton lock created with timeout={instance.timeout}, blocking={instance.blocking}," |
| 93 | f" cannot be changed to timeout={timeout}, blocking={blocking}" |
| 94 | ) |
| 95 | raise ValueError(msg) |
| 96 | return instance |
| 97 | |
| 98 | |
| 99 | class ReadWriteLock(metaclass=_ReadWriteLockMeta): |
nothing calls this directly
no outgoing calls
no test coverage detected
searching dependent graphs…