X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=src%2Fpyutils%2Ffiles%2Flockfile.py;h=0febca6bf5b754682f5e91f8604c70e566fd8a0d;hb=993b0992473c12294ed659e52b532e1c8cf9cd1e;hp=ee7346bf1042e0a0b85e36273c0ef8182c05f7b3;hpb=b38920f24d1ac948958480c540bc4b8436186765;p=pyutils.git

diff --git a/src/pyutils/files/lockfile.py b/src/pyutils/files/lockfile.py
index ee7346b..0febca6 100644
--- a/src/pyutils/files/lockfile.py
+++ b/src/pyutils/files/lockfile.py
@@ -2,7 +2,14 @@
 
 # © Copyright 2021-2022, Scott Gasch
 
-"""File-based locking helper."""
+"""This is a lockfile implementation I created for use with cronjobs
+on my machine to prevent multiple copies of a job from running in
+parallel.  When one job is running this code keeps a file on disk to
+indicate a lock is held.  Other copies will fail to start if they
+detect this lock until the lock is released.  There are provisions in
+the code for timing out locks, cleaning up a lock when a signal is
+received, gracefully retrying lock acquisition on failure, etc...
+"""
 
 from __future__ import annotations
 
@@ -91,11 +98,11 @@ class LockFile(contextlib.AbstractContextManager):
             signal.signal(signal.SIGTERM, self._signal)
         self.expiration_timestamp = expiration_timestamp
 
-    def locked(self):
+    def locked(self) -> bool:
         """Is it locked currently?"""
         return self.is_locked
 
-    def available(self):
+    def available(self) -> bool:
         """Is it available currently?"""
         return not os.path.exists(self.lockfile)
 
@@ -157,7 +164,7 @@ class LockFile(contextlib.AbstractContextManager):
             self._detect_stale_lockfile()
         return _try_acquire_lock_with_retries()
 
-    def release(self):
+    def release(self) -> None:
         """Release the lock"""
         try:
             os.unlink(self.lockfile)