Add ability to return immediately when a lock cannot be obtained inst…

src/filelock/__init__.py

@@ -11,7 +11,7 @@
 import warnings
 
 from ._api import AcquireReturnProxy, BaseFileLock
-from ._error import Timeout
+from ._error import ImmediateAquireError, Timeout
 from ._soft import SoftFileLock
 from ._unix import UnixFileLock, has_fcntl
 from ._windows import WindowsFileLock
@@ -45,4 +45,5 @@
     "WindowsFileLock",
     "BaseFileLock",
     "AcquireReturnProxy",
+    "ImmediateAquireError",
 ]

src/filelock/_api.py

@@ -10,7 +10,7 @@
 from types import TracebackType
 from typing import Any
 
-from ._error import Timeout
+from ._error import ImmediateAquireError, Timeout
 
 _LOGGER = logging.getLogger("filelock")
 
@@ -116,6 +116,7 @@ def acquire(
         poll_interval: float = 0.05,
         *,
         poll_intervall: float | None = None,
+        return_immediately=False,
     ) -> AcquireReturnProxy:
         """
         Try to acquire the file lock.
@@ -124,6 +125,8 @@ def acquire(
          if ``timeout < 0``, there is no timeout and this method will block until the lock could be acquired
         :param poll_interval: interval of trying to acquire the lock file
         :param poll_intervall: deprecated, kept for backwards compatibility, use ``poll_interval`` instead
+        :param return_immediately: defaults to False. If True, function will return immediately if it cannot obtain
+         a lock on the first attempt.
         :raises Timeout: if fails to acquire lock within the timeout period
         :return: a context object that will unlock the file when the context is exited
 
@@ -172,6 +175,9 @@ def acquire(
                 if self.is_locked:
                     _LOGGER.debug("Lock %s acquired on %s", lock_id, lock_filename)
                     break
+                elif return_immediately:
+                    _LOGGER.debug("Failed to immediately acquire lock %s on %s", lock_id, lock_filename)
+                    raise ImmediateAquireError(self._lock_file)
                 elif 0 <= timeout < time.monotonic() - start_time:
                     _LOGGER.debug("Timeout on acquiring lock %s on %s", lock_id, lock_filename)
                     raise Timeout(self._lock_file)

src/filelock/_error.py

@@ -12,6 +12,18 @@ def __str__(self) -> str:
         return f"The file lock '{self.lock_file}' could not be acquired."
 
 
+class ImmediateAquireError(TimeoutError):
+    """Raised when the lock could not be acquired immediately"""
+
+    def __init__(self, lock_file: str) -> None:
+        #: The path of the file lock.
+        self.lock_file = lock_file
+
+    def __str__(self) -> str:
+        return f"The file lock '{self.lock_file}' could not be acquired."
+
+
 __all__ = [
     "Timeout",
+    "ImmediateAquireError",
 ]

tests/test_filelock.py

@@ -14,7 +14,7 @@
 import pytest
 from _pytest.logging import LogCaptureFixture
 
-from filelock import BaseFileLock, FileLock, SoftFileLock, Timeout, UnixFileLock, WindowsFileLock
+from filelock import BaseFileLock, FileLock, ImmediateAquireError, SoftFileLock, Timeout, UnixFileLock, WindowsFileLock
 
 
 @pytest.mark.parametrize(
@@ -259,6 +259,29 @@ def test_timeout(lock_type: type[BaseFileLock], tmp_path: Path) -> None:
     assert not lock_2.is_locked
 
 
+@pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
+def test_return_immediately(lock_type: type[BaseFileLock], tmp_path: Path) -> None:
+    # raises ImmediateAquireError error when the lock cannot be acquired
+    lock_path = tmp_path / "a"
+    lock_1, lock_2 = lock_type(str(lock_path)), lock_type(str(lock_path))
+
+    # acquire lock 1
+    lock_1.acquire()
+    assert lock_1.is_locked
+    assert not lock_2.is_locked
+
+    # try to acquire lock 2
+    with pytest.raises(ImmediateAquireError, match="The file lock '.*' could not be acquired."):
+        lock_2.acquire(return_immediately=True)
+    assert not lock_2.is_locked
+    assert lock_1.is_locked
+
+    # release lock 1
+    lock_1.release()
+    assert not lock_1.is_locked
+    assert not lock_2.is_locked
+
+
 @pytest.mark.parametrize("lock_type", [FileLock, SoftFileLock])
 def test_default_timeout(lock_type: type[BaseFileLock], tmp_path: Path) -> None:
     # test if the default timeout parameter works