Initial commit

Author: z3rone

.github/workflows/python-publish.yml

@@ -0,0 +1,70 @@
+# This workflow will upload a Python Package to PyPI when a release is created
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries
+
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+name: Upload Python Package
+
+on:
+  release:
+    types: [published]
+
+permissions:
+  contents: read
+
+jobs:
+  release-build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.x"
+
+      - name: Build release distributions
+        run: |
+          # NOTE: put your own distribution build steps here.
+          python -m pip install build
+          python -m build
+
+      - name: Upload distributions
+        uses: actions/upload-artifact@v4
+        with:
+          name: release-dists
+          path: dist/
+
+  pypi-publish:
+    runs-on: ubuntu-latest
+    needs:
+      - release-build
+    permissions:
+      # IMPORTANT: this permission is mandatory for trusted publishing
+      id-token: write
+
+    # Dedicated environments with protections for publishing are strongly recommended.
+    # For more information, see: https://docs.github.com/en/actions/deployment/targeting-different-environments/using-environments-for-deployment#deployment-protection-rules
+    environment:
+      name: pypi
+      # OPTIONAL: uncomment and update to include your PyPI project URL in the deployment status:
+      # url: https://pypi.org/p/YOURPROJECT
+      #
+      # ALTERNATIVE: if your GitHub Release name is the PyPI project version string
+      # ALTERNATIVE: exactly, uncomment the following line instead:
+      # url: https://pypi.org/project/YOURPROJECT/${{ github.event.release.name }}
+
+    steps:
+      - name: Retrieve release distributions
+        uses: actions/download-artifact@v4
+        with:
+          name: release-dists
+          path: dist/
+
+      - name: Publish release distributions to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          packages-dir: dist/

.github/workflows/python-test.yml

@@ -0,0 +1,35 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Run Tests
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install pytest
+        python -m pip install .
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Test with pytest
+      run: |
+        pytest

README.md

@@ -0,0 +1,75 @@
+# Dirlock
+
+[![PyPi badge](https://img.shields.io/pypi/v/dirlock)](https://pypi.org/project/dirlock/)
+![pytest badge](https://github.com/z3rone-org/dirlock/actions/workflows/python-test.yml/badge.svg)
+
+A simple directory-based lock implementation for Python. This package provides a lightweight and effective way
+to coordinate access to shared resources using a lock directory mechanism.
+This does not require any file locking capabilities of the underlying filesystem or network share.
+
+## Installation
+
+Install the package via pip:
+
+```bash
+pip install dirlock
+```
+
+## Usage
+
+You can use the `DirLock` class to acquire and release locks in your Python code. The class also supports usage within a `with` clause for convenience.
+
+### Example: Using the Lock Explicitly
+
+```python
+import time
+from dirlock import DirLock
+
+lock_dir_path = "/tmp/mylockdir.lock"
+lock = DirLock(lock_dir_path)
+
+print("Acquire lock...")
+lock.acquire()
+print("Lock acquired!")
+
+# Perform critical section tasks here
+# Simulating work
+time.sleep(5)
+
+# Release the lock
+lock.release()
+print("Lock released.")
+```
+
+### Example: Using the Lock in a `with` Clause
+
+```python
+import time
+from dirlock import DirLock
+
+lock_dir_path = "/tmp/mylockdir.lock"
+
+with DirLock(lock_dir_path) as lock:
+    print("Lock acquired!")
+    # Perform critical section tasks
+    time.sleep(5)  # Simulate work
+    print("Work done!")
+
+# The lock is automatically released when the block exits.
+print("Lock released.")
+```
+
+## Parameters
+
+- `lock_dir` (str): The path to the lock directory.
+- `retry_interval` (float, default=0.1): Time to wait before retrying if the lock cannot be acquired.
+- `timeout_interval` (float, default=-1): Timeout for acquiring lock. Set to negative value for no timeout.
+
+## Change Default Values
+You can change the default value for  `retry_interval`
+via `DirLock.retry_interval=<new_value>`.
+
+## License
+
+This package is licensed under the MIT License.
+

dirlock/__init__.py

@@ -0,0 +1,75 @@
+import os
+import time
+from datetime import datetime
+
+
+class DirLock:
+    default_retry_interval: float = 0.1
+    def __init__(self,
+                 lock_dir: str,
+                 retry_interval: float = None,
+                 timeout_interval: float = -1):
+        """
+        Initialize the DirLock.
+
+        Args:
+            lock_dir (str): Path to the lock directory.
+            retry_interval (float): Time to wait before retrying (in seconds).
+            timeout_interval (float): Timeout
+        """
+        self.lock_dir = str(lock_dir)
+        self.timeout_interval = timeout_interval
+
+        if retry_interval is None:
+            self.retry_interval = DirLock.default_retry_interval
+        else:
+            self.retry_interval = retry_interval
+        self.acquired = False
+
+    def acquire(self):
+        """
+        Acquire the lock by following the directory-based lock mechanism.
+        """
+        start_time = datetime.now()
+
+        while True:
+            try:
+                os.mkdir(self.lock_dir)
+                self.acquired = True
+                break
+            except FileExistsError:
+                pass
+
+            if self.timeout_interval >= 0.0:
+                if (datetime.now() - start_time).total_seconds() > self.timeout_interval:
+                    raise LockTimeoutException()
+
+            # If the lock directory exists, retry
+            time.sleep(self.retry_interval)
+
+    def release(self):
+        """
+        Release the lock if it is held by this instance.
+        """
+        try:
+            os.rmdir(self.lock_dir)
+        except FileNotFoundError:
+            pass
+        self.acquired = False
+
+    def __enter__(self):
+        """
+        Context management entry point.
+        """
+        self.acquire()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """
+        Context management exit point.
+        """
+        self.release()
+
+class LockTimeoutException(Exception):
+    def __init__(self, message="acquiring lock timed out"):
+        super().__init__(message)

setup.py

@@ -0,0 +1,27 @@
+from setuptools import setup, find_packages
+
+setup(
+    name="dirlock",
+    description="A simple directory based lock implementation for Python.",
+    use_scm_version=True,
+    long_description=open("README.md", encoding="utf-8").read(),
+    long_description_content_type="text/markdown",
+    author="Falk B. Schimweg",
+    author_email="git@falk.schimweg.de",
+    url="https://github.com/z3rone-org/dirlock",
+    license="MIT",
+    packages=find_packages(),
+    classifiers=[
+        "Programming Language :: Python :: 3",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: OS Independent",
+    ],
+    extras_require={
+        "dev": [
+            "pytest",
+        ]
+    },
+    setup_requires=['setuptools_scm'],
+    python_requires=">=3.8",
+)
+

tests/test_lock.py

@@ -0,0 +1,61 @@
+from datetime import datetime
+import os
+from dirlock import DirLock, LockTimeoutException
+
+
+def test_two_locks(tmpdir):
+    lockdir = os.path.join(tmpdir, '.lock')
+    lock1 = DirLock(lockdir)
+    lock2 = DirLock(lockdir)
+
+    assert not lock1.acquired
+    assert not lock2.acquired
+
+    # Acquire the lock
+    lock1.acquire()
+
+    assert lock1.acquired
+    assert not lock2.acquired
+
+    lock1.release()
+    lock2.acquire()
+
+    assert not lock1.acquired
+    assert lock2.acquired
+
+    lock2.release()
+
+    assert not lock1.acquired
+    assert not lock2.acquired
+
+
+def test_timeout(tmpdir):
+    lockdir = os.path.join(tmpdir, '.lock')
+    lock1 = DirLock(lockdir)
+    lock2 = DirLock(lockdir, timeout_interval=3)
+
+    lock1.acquire()
+    assert lock1.acquired
+
+    try:
+        lock2.acquire()
+        assert False
+    except LockTimeoutException:
+        assert not lock2.acquired
+
+    lock1.release()
+    assert not lock1.acquired
+    assert not lock2.acquired
+
+
+def test_context_manager(tmpdir):
+    lockdir = os.path.join(tmpdir, '.lock')
+    lock = DirLock(str(lockdir))
+
+    # Use the context manager to acquire and release the lock
+    with lock:
+        # Check that the lock directory was created with the correct UUID
+        assert os.path.exists(str(lockdir))
+
+    # Ensure that the lock directory is removed after exiting the context
+    assert not os.path.exists(str(lockdir))