Simple lockfile

lockfile.py

from __future__ import print_function

import os
import time
import errno
import socket

class FileLock(object):

    CONTENTS = "Locked by\n * host: {}\n * process: {}".format(
        socket.gethostname(),
        os.getpid(),
    ).encode("ascii")

    def __init__(self, path, poll=0.2):
        self.__path = path
        self.__poll = poll
        self.__file_handle = None

    def path(self):
        return self.__path

    def locked(self):
        return self.__file_handle is not None

    def acquire(self, blocking=True, timeout=-1):
        if self.locked():
            raise RuntimeError("Already locked")
        if not blocking:
            return self.__acquire_lock()
        start_time = current_time = time.time()
        while timeout < 0 or current_time - start_time < timeout:
            if self.__acquire_lock():
                return True
            time.sleep(self.__poll)
            current_time = time.time()
        return False

    def release(self):
        if not self.locked():
            raise RuntimeError("Already unlocked")
        os.close(self.__file_handle)
        self.__file_handle = None
        os.remove(self.__path)

    def __acquire_lock(self):
        flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY
        try:
            self.__file_handle = os.open(self.__path, flags)
        except OSError as err:
            self.__file_handle = None
            if err.errno != errno.EEXIST:
                raise
            return False
        else:
            os.write(self.__file_handle, self.CONTENTS)
        return True

    def __enter__(self):
        self.acquire()
        return self

    def __exit__(self, *_):
        self.release()


if __name__ == "__main__":
    import tempfile
    path = os.path.join(tempfile.mkdtemp(), "lockfile")
    with FileLock(path) as lock:
        assert lock.locked()
        assert not FileLock(path).acquire(blocking=False)
        assert not FileLock(path).acquire(timeout=0.2)