Install & Compatibility
Where this runs
tested against v1.0.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.95 runs
installs and imports cleanly · install 0.0s · import 0.004s · 17.8MB
glibcpy 3.10–3.95 runs
installs and imports cleanly · install 1.5s · import 0.002s · 18MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
locket
✓ import locket
The primary interface is through the `locket` module directly, often using `locket.lock_file()`.
LockError
✓ from locket import LockError
The specific exception raised by locket operations.
This quickstart demonstrates how to use `locket.lock_file()` with a context manager for safe and automatic lock acquisition and release. It also shows the explicit `acquire()` and `release()` pattern, which requires careful handling to prevent deadlocks or `LockError` if `release()` is called on an unlocked lock. The `timeout` parameter prevents indefinite blocking.
import locket
import os
import time
# Define a lock file path, using an environment variable for flexibility
lock_file_path = os.environ.get('LOCKET_EXAMPLE_PATH', './my_app.lock')
print(f"Attempting to acquire lock on {lock_file_path}...")
try:
# Acquire the lock using a context manager, with a 5-second timeout
# If another process holds the lock, this will block for up to 5 seconds.
# If the lock cannot be acquired, a LockError is raised.
with locket.lock_file(lock_file_path, timeout=5):
print(f"Lock acquired on {lock_file_path}. Performing critical action...")
# Simulate work within the critical section
time.sleep(2)
print("Critical action complete. Lock will be released automatically.")
except locket.LockError:
print(f"Could not acquire lock on {lock_file_path} within the timeout. Another process might hold it.")
except Exception as e:
print(f"An unexpected error occurred: {e}")
print("\nDemonstrating explicit acquire/release (less recommended)...")
lock = None
try:
# Acquire the lock explicitly
lock = locket.lock_file(lock_file_path, timeout=1)
lock.acquire()
print(f"Lock acquired on {lock_file_path} (manual). Performing critical action...")
time.sleep(1)
print("Critical action complete (manual).")
finally:
# Always ensure the lock is released, even if errors occur
if lock:
try:
lock.release()
print("Lock released (manual).")
except locket.LockError:
print("Warning: Attempted to release an already unlocked lock. This should not happen if `release()` is paired with `acquire()`.")
Debug
Known issues
gotchaLocket provides non-reentrant locks. This means a thread cannot acquire a lock it already holds without blocking or raising a `LockError` if a timeout is specified. If you need reentrant behavior, use `threading.RLock` or ensure your code structure avoids re-acquiring the same lock.fixAvoid recursive lock acquisition, or use `threading.RLock` if reentrant locking is strictly required within a single process.
affects: All versions
breakingThe behavior of Locket's file-based locks after a process `fork` is undefined. This means using `locket` directly with `multiprocessing` modules that rely on forking might lead to unpredictable results or deadlocks.fixAvoid inheriting locks across `fork` calls. Acquire locks *after* forking in each child process if separate locking is needed, or use `multiprocessing.Lock` for IPC within the `multiprocessing` module.
affects: All versions
gotchaCalling `release()` on an already unlocked lock will raise a `locket.LockError`. This typically indicates a logic error where `release()` is called more times than `acquire()` or when a lock is not held by the current process/thread.fixAlways pair `acquire()` with `release()`, ideally using a `try...finally` block or, preferably, the `with` statement (context manager) provided by `locket.lock_file()` to ensure automatic and safe release.
affects: All versions
deprecatedThe last release of Locket (version 1.0.0) was in April 2022. While functional, the library has not received recent updates, which might imply a maintenance-only status. Users should be aware that it may not receive updates for newer Python versions or address new edge cases/bugs promptly.fixEvaluate for your specific use case. For critical applications requiring active maintenance or features, consider alternatives like `fasteners` or `oslo.concurrency` (though `oslo.concurrency` might be more heavyweight for simple needs).
affects: Post 1.0.0 development (lack of new versions)
Errors
Common errors & fixes
locket.LockError: attempted to release an unlocked lock
This error occurs when `release()` is called on a `locket` lock that is not currently held by the calling process/thread, or if `release()` is called more times than `acquire()`.
fixEnsure `acquire()` and `release()` calls are properly paired, ideally by using the `with` statement for automatic and safe lock management. Example: `with locket.lock_file('path/to/lock/file'): # critical section` locket.LockError: could not acquire lock
This error is raised when `locket.lock_file()` is called with a `timeout` parameter (including `timeout=0` for non-blocking) and the lock cannot be acquired within the specified time because another process holds it.
fixHandle the `locket.LockError` in a `try...except` block to manage scenarios where the lock cannot be obtained immediately or within the timeout. You might retry after a delay or log the contention. Example: `try: with locket.lock_file('path/to/lock/file', timeout=5): # critical section except locket.LockError: print('Failed to acquire lock.')` ModuleNotFoundError: No module named 'locket'
This error means the `locket` library is not installed in the Python environment where the code is being executed.
fixInstall the `locket` library using pip. Command: `pip install locket`
Locket provides non-reentrant locks. This means a thread cannot acquire a lock it already holds without blocking or raising a `LockError` if a timeout is specified.
The `locket` library implements non-reentrant locks, meaning the same thread cannot acquire the same lock more than once without explicitly releasing it first. Attempting to do so will lead to blocking or a `LockError` if a timeout is set.
fixRefactor the code to avoid recursive lock acquisition. If reentrant behavior is genuinely required within a single process/thread, consider using `threading.RLock` instead of `locket`'s file-based locks for that specific critical section.
The behavior of Locket's file-based locks after a process `fork` is undefined.
When a process using `locket`'s file-based locks forks, the inherited lock state in the child process is undefined. This can lead to deadlocks or unpredictable behavior when both parent and child processes try to interact with the same lock.
fixAvoid inheriting `locket` locks across `fork` calls. Acquire locks *after* forking in each child process if separate locking is needed, or use `multiprocessing.Lock` for inter-process communication specifically designed for use with `multiprocessing`.
Upgrade
Version history
1.0.0latest on PyPI · released Apr 20, 2022
Audit
Dependencies
pythonrequiredRequired runtime environment.