Thread lock until a file **cannot** be locked
| 42 | |
| 43 | |
| 44 | class FileFreeLock(BaseFileLock): |
| 45 | """Thread lock until a file **cannot** be locked""" |
| 46 | |
| 47 | def __init__(self, lock_file, *args, **kwargs): |
| 48 | self.filelock = FileLock(lock_file) |
| 49 | super().__init__(lock_file, *args, **kwargs) |
| 50 | self._lock_file_fd = None |
| 51 | |
| 52 | def _acquire(self): |
| 53 | try: |
| 54 | self.filelock.acquire(timeout=0.01, poll_interval=0.02) # Try to lock once |
| 55 | except Timeout: |
| 56 | # We couldn't acquire the lock, the file is locked! |
| 57 | self._lock_file_fd = self.filelock.lock_file |
| 58 | else: |
| 59 | # We were able to acquire the lock, the file is not yet locked! |
| 60 | self.filelock.release() |
| 61 | self._lock_file_fd = None |
| 62 | |
| 63 | def _release(self): |
| 64 | self._lock_file_fd = None |
| 65 | |
| 66 | @property |
| 67 | def is_locked(self) -> bool: |
| 68 | return self._lock_file_fd is not None |
| 69 | |
| 70 | |
| 71 | # lists - summarize long lists similarly to NumPy |
no outgoing calls
no test coverage detected
searching dependent graphs…