zc.lockfile
The zc.lockfile package provides a basic, portable implementation of interprocess locks using file-locking primitives. It is not specifically for locking files themselves, but rather for providing general-purpose interprocess locks with a file-based mechanism. Currently at version 4.0, the library is actively maintained with updates primarily focused on Python version compatibility and bug fixes.
Warnings
- breaking In version 2.0, `zc.lockfile.LockFile` changed from an old-style to a new-style class. Additionally, `SimpleLockFile` was extracted, altering implicit behavior for writing to the lock file. Code relying on old-style class introspection or the previous implicit lock file writing mechanism will break.
- breaking Version 4.0 replaced the `pkg_resources` namespace with PEP 420 native namespace packaging. This change affects how the package is discovered and imported in certain environments, particularly complex Zope setups or systems with custom package loaders.
- gotcha Failing to call `lock.close()` will leave the lock file on the filesystem. This prevents other processes from acquiring the lock indefinitely, even if the original holding process terminates or crashes, leading to resource leaks or deadlocks.
- gotcha The `lock.close()` method releases the lock but *does not remove the lock file* from the filesystem. The file persists by design. If the lock file is no longer needed after the lock is released, it must be explicitly removed.
- gotcha In containerized environments (e.g., Docker), process IDs (PIDs) can be misleading or identical across different containers on the same host. Relying solely on PIDs within lock files for debugging or identification can be problematic.
Install
-
pip install zc.lockfile
Imports
- LockFile
from zc.lockfile import LockFile
- LockError
from zc.lockfile import LockError
Quickstart
import zc.lockfile
import os
import time
lock_file_path = "my_app.lock"
lock = None
try:
# Attempt to acquire the lock
lock = zc.lockfile.LockFile(lock_file_path)
print(f"Process {os.getpid()}: Acquired lock on {lock_file_path}")
# Simulate critical section work
print(f"Process {os.getpid()}: Doing some critical work...")
time.sleep(2) # Simulate work
print(f"Process {os.getpid()}: Critical work complete.")
except zc.lockfile.LockError:
print(f"Process {os.getpid()}: Could not acquire lock on {lock_file_path}. Another process holds it.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
finally:
if lock:
lock.close()
print(f"Process {os.getpid()}: Released lock on {lock_file_path}")
# Optionally remove the lock file if it should not persist
# In a real application, consider if other processes might still need to check for its existence
try:
os.remove(lock_file_path)
print(f"Process {os.getpid()}: Removed lock file {lock_file_path}")
except OSError as e:
print(f"Process {os.getpid()}: Error removing lock file {lock_file_path}: {e}")