add flockd and test; user can choose whether blocking and create lock…

… file

.gitignore

@@ -11,3 +11,5 @@ __pycache__/
 # due to using tox and pytest
 .tox
 .cache
+env/
+venv/

src/shinny_filelock/__init__.py

@@ -0,0 +1 @@
+from ._flockd import flocked

src/shinny_filelock/_flockd.py

@@ -0,0 +1,31 @@
+import fcntl
+import os
+from contextlib import contextmanager
+
+
+@contextmanager
+def flocked(path, blocking=False, create_file=False):
+    """
+    :param path: file path
+    :param blocking: blocking lock
+    :param create_file: create file if not exists
+    """
+    fd = -1
+    try:
+        fd_flags = os.O_RDONLY | os.O_NOCTTY
+        if create_file:
+            fd_flags |= os.O_CREAT
+        fd = os.open(path, fd_flags)
+        if fd == -1:
+            raise ValueError()
+        try:
+            flags = fcntl.LOCK_EX
+            if not blocking:
+                flags |= fcntl.LOCK_NB
+            fcntl.flock(fd, flags)
+            yield
+        finally:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+    finally:
+        if fd != -1:
+            os.close(fd)

tests/_timeout.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+
+import signal
+from contextlib import ContextDecorator
+
+
+def raise_timeout(signum, frame):
+    raise TimeoutError()
+
+
+class timeout(ContextDecorator):
+    """Raises TimeoutError when the gien time in seconds elapsed.
+    """
+
+    def __init__(self, seconds):
+        self._seconds = seconds
+
+    def __enter__(self):
+        if self._seconds:
+            self._replace_alarm_handler()
+            signal.setitimer(signal.ITIMER_REAL, self._seconds)
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        if self._seconds:
+            self._restore_alarm_handler()
+            signal.alarm(0)
+
+    def _replace_alarm_handler(self):
+        self._old_alarm_handler = signal.signal(signal.SIGALRM,
+                                                raise_timeout)
+
+    def _restore_alarm_handler(self):
+        signal.signal(signal.SIGALRM, self._old_alarm_handler)

tests/test_flock.py

@@ -0,0 +1,43 @@
+# the inclusion of the tests module is not meant to offer best practices for
+# testing in general, but rather to support the `find_packages` example in
+# setup.py that excludes installing the "tests" package
+import unittest
+
+from shinny_filelock import flocked
+
+from ._timeout import timeout
+
+
+class TestSimple(unittest.TestCase):
+
+    def test_non_blocking_lock(self):
+        local_path = "/tmp/test.lock"
+        with flocked(local_path):
+            try:
+                with flocked(local_path):
+                    pass
+            except Exception as e:
+                self.assertEqual(e.__class__, BlockingIOError)
+
+    def test_blocking_lock(self):
+        local_path = "/tmp/test-block.lock"
+        with flocked(local_path, blocking=True):
+            try:
+                with timeout(2):
+                    with flocked(local_path, blocking=True):
+                        pass
+            except Exception as e:
+                self.assertEqual(e.__class__, TimeoutError)
+
+    def test_blocking_with_non_blocking_lock(self):
+        local_path = "/tmp/test-mix.lock"
+        with flocked(local_path, blocking=True):
+            try:
+                with flocked(local_path, blocking=False):
+                    pass
+            except Exception as e:
+                self.assertEqual(e.__class__, BlockingIOError)
+
+
+if __name__ == '__main__':
+    unittest.main()