diff --git a/src/filelock/_api.py b/src/filelock/_api.py index 282106fd..b7cc7121 100644 --- a/src/filelock/_api.py +++ b/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) diff --git a/tests/test_filelock.py b/tests/test_filelock.py index f439ff6d..2b212a4b 100644 --- a/tests/test_filelock.py +++ b/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