philiprehberger-lock-run
v0.2.0529 downloadsMIT
Prevent duplicate script execution using file-based locking
$
pip install philiprehberger-lock-runREADME
philiprehberger-lock-run
Prevent duplicate script execution using file-based locking.
Installation
pip install philiprehberger-lock-run
Usage
Context Manager
from philiprehberger_lock_run import lock
with lock("my-job"):
do_work()
Decorator
from philiprehberger_lock_run import locked
@locked("my-job")
def scheduled_task():
...
Timeout
Wait up to 10 seconds for the lock to become available:
with lock("my-job", timeout=10):
do_work()
Custom Lock Directory
with lock("my-job", lock_dir="/var/lock"):
do_work()
Inspecting and cleaning up locks
Check whether another process currently holds a lock (non-blocking):
from philiprehberger_lock_run import is_locked
if is_locked("my-job"):
print("Another process is running my-job")
Remove orphaned lock files older than a threshold (defaults to 24 hours). Only files that are not currently held are removed:
from philiprehberger_lock_run import cleanup_locks
removed = cleanup_locks(max_age_seconds=3600)
print(f"Removed {len(removed)} stale lock files")
API
| Name | Description |
|---|---|
lock(name, *, timeout=0, lock_dir=None) | Context manager that acquires a file lock. Raises LockError on failure. |
locked(name, **kwargs) | Decorator that wraps the function body in a file lock. |
is_locked(name, *, lock_dir=None) | Return True if another process currently holds the named lock. Performs a non-blocking try-acquire. |
cleanup_locks(lock_dir=None, *, max_age_seconds=86400) | Remove orphaned lock files older than max_age_seconds that are not currently held. Returns the list of removed file names. |
LockError | Raised when a lock cannot be acquired. Extends RuntimeError. |
Development
pip install -e .
python -m pytest tests/ -v
Support
If you find this project useful: