"""Class to lock files"""
import os
import errno
import sys
import time
import logging
_log = logging.getLogger(__name__)
# pylint: disable=E0401,W0201,C0301
[docs]class FileLock(object):
"""A file locking mechanism that has context-manager support so
you can use it in a ``with`` statement. This should be relatively cross
compatible as it doesn't rely on ``msvcrt`` or ``fcntl`` for the locking.
"""
[docs] class FileLockException(Exception):
"""Exception to the file lock object"""
pass
def __init__(self, protected_file_path, timeout=None, delay=1, lock_file_contents=None):
"""
Prepare the file locker. Specify the file to lock and optionally
the maximum timeout and the delay between each attempt to lock.
"""
self.is_locked = False
self.lockfile = protected_file_path + '.lock'
self.timeout = timeout
self.delay = delay
self._lock_file_contents = lock_file_contents
if self._lock_file_contents is None:
self._lock_file_contents = 'Owning process args:\n'
for arg in sys.argv:
self._lock_file_contents += arg + '\n'
[docs] def locked(self):
"""
Returns True iff the file is owned by THIS FileLock instance.
(Even if this returns false, the file could be owned by another FileLock instance,
possibly in a different thread or process).
"""
return self.is_locked
[docs] def available(self):
"""
Returns True iff the file is currently available to be locked.
"""
return not os.path.exists(self.lockfile)
[docs] def lock_exists(self):
"""
Returns True iff the external lockfile exists.
"""
return os.path.exists(self.lockfile)
[docs] def acquire(self, blocking=True):
"""
Acquire the lock, if possible. If the lock is in use, and `blocking` is False,
return False. Otherwise, check again every `self.delay` seconds until it either gets the
lock or exceeds `timeout` number of seconds, in which case it raises an exception.
"""
start_time = time.time()
while True:
try:
# Attempt to create the lockfile.
# These flags cause os.open to raise an OSError if the file already exists.
fd = os.open(self.lockfile, os.O_CREAT | os.O_EXCL | os.O_RDWR)
with os.fdopen(fd, 'a') as f:
# Print some info about the current process as debug info
# for anyone who bothers to look.
f.write(self._lock_file_contents)
break
except OSError as e:
if e.errno != errno.EEXIST:
raise
if self.timeout is not None and (time.time() - start_time) >= self.timeout:
raise FileLock.FileLockException('Timeout occurred.')
if not blocking:
return False
time.sleep(self.delay)
self.is_locked = True
return True
[docs] def release(self):
"""
Get rid of the lock by deleting the lockfile.
When working in a `with` statement, this gets automatically
called at the end.
"""
self.is_locked = False
os.unlink(self.lockfile)
def __enter__(self):
"""
Activated when used in the with statement.
Should automatically acquire a lock to be used in the with block.
"""
self.acquire()
return self
def __exit__(self, type, value, traceback):
"""
Activated at the end of the with statement.
It automatically releases the lock if it isn't locked.
"""
self.release()
def __del__(self):
"""
Make sure this ``FileLock`` instance doesn't leave a .lock file
lying around.
"""
if self.is_locked:
self.release()
[docs] def purge(self):
"""
For debug purposes only. Removes the lock file from the hard disk.
"""
if os.path.exists(self.lockfile):
self.release()
return True
return False