(
cls,
lock_file: str | os.PathLike[str],
timeout: float = -1,
*,
blocking: bool = True,
is_singleton: bool = True,
)
| 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