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

src/filelock/_api.py

@@ -116,6 +116,7 @@ def acquire(
         poll_interval: float = 0.05,
         *,
         poll_intervall: float | None = None,
+        blocking: bool = True,
     ) -> 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 blocking: defaults to True. If False, function will return immediately if it cannot obtain a lock on the
+         first attempt. Otherwise this method will block until the timeout expires or the lock is acquired.
         :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 blocking is False:
+                    _LOGGER.debug("Failed to immediately acquire lock %s on %s", lock_id, lock_filename)
+                    raise Timeout(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)

tests/test_filelock.py

@@ -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_non_blocking(lock_type: type[BaseFileLock], tmp_path: Path) -> None:
+    # raises Timeout 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(Timeout, match="The file lock '.*' could not be acquired."):
+        lock_2.acquire(blocking=False)
+    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