flufl-lock

import tempfile import os from flufl.lock import Lock, AlreadyLockedError # Create a temporary file to use as the lock file # In a real application, use a persistent path accessible to all processes with tempfile.NamedTemporaryFile(delete=False) as tmp_lock_file: lock_file_path = tmp_lock_file.name try: # Acquire the lock using a context manager for automatic release # default lifetime is 15 seconds with Lock(lock_file_path) as lock: print(f"Lock acquired on {lock_file_path}") print(f"Is locked: {lock.is_locked}") print("Performing some locked operation...") # Simulate another process trying to acquire the same lock try: with Lock(lock_file_path, default_timeout=1) as other_lock: print("This should not be reached if the lock is held.") except AlreadyLockedError: print("Another process correctly detected the lock is held.") # The lock is automatically released when exiting the 'with' block print("Lock released.") # Acquire and release manually lock = Lock(lock_file_path) lock.lock() print("Lock acquired manually.") lock.refresh() print("Lock refreshed (lifetime extended).") lock.unlock() print("Lock released manually.") finally: # Clean up the temporary lock file if os.path.exists(lock_file_path): os.remove(lock_file_path) # The library might create a claim file; ensure it's also cleaned. # For this simple example, we assume lock_file_path is sufficient.