diff --git a/.gitignore b/.gitignore index e81f084..1421f20 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,5 @@ __pycache__/ # due to using tox and pytest .tox .cache +env/ +venv/ diff --git a/src/sample/__init__.py b/src/sample/__init__.py deleted file mode 100644 index c8f1064..0000000 --- a/src/sample/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -def main(): - """Entry point for the application script""" - print("Call your main application code here") diff --git a/src/sample/simple.py b/src/sample/simple.py deleted file mode 100644 index c929f88..0000000 --- a/src/sample/simple.py +++ /dev/null @@ -1,2 +0,0 @@ -def add_one(number): - return number + 1 diff --git a/src/shinny_filelock/__init__.py b/src/shinny_filelock/__init__.py new file mode 100644 index 0000000..de412b6 --- /dev/null +++ b/src/shinny_filelock/__init__.py @@ -0,0 +1 @@ +from ._flockd import flocked diff --git a/src/shinny_filelock/_flockd.py b/src/shinny_filelock/_flockd.py new file mode 100644 index 0000000..3ad77aa --- /dev/null +++ b/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) diff --git a/tests/_timeout.py b/tests/_timeout.py new file mode 100644 index 0000000..846ff4e --- /dev/null +++ b/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) diff --git a/tests/test_flock.py b/tests/test_flock.py new file mode 100644 index 0000000..cb9f0f6 --- /dev/null +++ b/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() diff --git a/tests/test_simple.py b/tests/test_simple.py deleted file mode 100644 index b7a0ee3..0000000 --- a/tests/test_simple.py +++ /dev/null @@ -1,17 +0,0 @@ -# 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 sample.simple import add_one - - -class TestSimple(unittest.TestCase): - - def test_add_one(self): - self.assertEqual(add_one(5), 6) - - -if __name__ == '__main__': - unittest.main()