philiprehberger-lock-run 0.2.0

Prevent duplicate script execution using file-based locking

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:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT