0.0.22

.gitignore

@@ -12,3 +12,4 @@ html
 .coverage
 htmlcov
 test.py
+CLAUDE.md

README.md

@@ -28,6 +28,7 @@ It adds several useful features to Python’s standard synchronization primitive
 
 - [**Installation**](#installation)
 - [**Lock protocols**](#lock-protocols)
+- [**Empty locks**](#empty-locks)
 - [**`SmartLock` turns deadlocks into exceptions**](#smartlock-turns-deadlocks-into-exceptions)
 - [**Test your locks**](#test-your-locks)
 
@@ -114,6 +115,32 @@ print(isinstance(Lock(), AsyncContextLockProtocol)) # True
 If you use type hints and static verification tools like [mypy](https://github.com/python/mypy), we highly recommend using the narrowest applicable protocol for your use case.
 
 
+## Empty locks
+
+Sometimes a piece of code expects a lock, but in a particular case no synchronization is actually needed. Instead of branching on whether to lock, you can inject a lock that does nothing. `locklib` provides two such no-op locks: `EmptyLock` and its asynchronous counterpart `AsyncEmptyLock`. Their `acquire`/`release` methods and context-manager forms return immediately and never block:
+
+```python
+from locklib import EmptyLock
+
+lock = EmptyLock()
+
+with lock:
+    ...  # nothing is actually locked
+```
+
+```python
+from locklib import AsyncEmptyLock
+
+lock = AsyncEmptyLock()
+
+async def function():
+    async with lock:
+        ...  # nothing is actually locked
+```
+
+`EmptyLock` implements `ContextLockProtocol` and `AsyncEmptyLock` implements `AsyncContextLockProtocol` (and both implement `LockProtocol`), so each one is a drop-in substitute wherever the corresponding protocol is expected.
+
+
 ## `SmartLock` turns deadlocks into exceptions
 
 `locklib` includes a lock that prevents [deadlocks](https://en.wikipedia.org/wiki/Deadlock) — `SmartLock`, based on [Wait-for Graph](https://en.wikipedia.org/wiki/Wait-for_graph). You can use it like a regular [`Lock` from the standard library](https://docs.python.org/3/library/threading.html#lock-objects). Let’s verify that it prevents [race conditions](https://en.wikipedia.org/wiki/Race_condition) in the same way:

locklib/__init__.py

@@ -5,6 +5,10 @@
 from locklib.errors import (
     ThereWasNoSuchEventError as ThereWasNoSuchEventError,
 )
+from locklib.locks.empty.async_empty_lock import (
+    AsyncEmptyLock as AsyncEmptyLock,
+)
+from locklib.locks.empty.empty_lock import EmptyLock as EmptyLock
 from locklib.locks.smart_lock.lock import SmartLock as SmartLock
 from locklib.locks.tracer.tracer import (
     LockTraceWrapper as LockTraceWrapper,

locklib/locks/empty/async_empty_lock.py

@@ -0,0 +1,24 @@
+from types import TracebackType
+from typing import Optional, Type
+
+
+class AsyncEmptyLock:
+    """Provide the async-context-lock interface while deliberately doing no locking.
+
+    The asynchronous counterpart of ``EmptyLock``: it mirrors the shape of
+    ``asyncio.Lock`` (an awaitable ``acquire`` and a synchronous ``release``)
+    but performs no synchronization. It is stateless, so it never blocks and
+    can be reused freely.
+    """
+
+    async def __aenter__(self) -> None:
+        await self.acquire()
+
+    async def __aexit__(self, exception_type: Optional[Type[BaseException]], exception_value: Optional[BaseException], traceback: Optional[TracebackType]) -> None:
+        self.release()
+
+    async def acquire(self) -> None:
+        ...
+
+    def release(self) -> None:
+        ...

locklib/locks/empty/empty_lock.py

@@ -0,0 +1,23 @@
+from types import TracebackType
+from typing import Optional, Type
+
+
+class EmptyLock:
+    """Provide the context-lock interface while deliberately doing no locking.
+
+    Useful when some code expects a lock but no synchronization is actually
+    needed, so a no-op lock can be injected instead of branching on whether to
+    lock. It is stateless, so it never blocks and can be reused freely.
+    """
+
+    def __enter__(self) -> None:
+        self.acquire()
+
+    def __exit__(self, exception_type: Optional[Type[BaseException]], exception_value: Optional[BaseException], traceback: Optional[TracebackType]) -> None:
+        self.release()
+
+    def acquire(self) -> None:
+        ...
+
+    def release(self) -> None:
+        ...

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = 'setuptools.build_meta'
 
 [project]
 name = 'locklib'
-version = '0.0.21'
+version = '0.0.22'
 authors = [
   { name='Evgeniy Blinov', email='zheni-b@yandex.ru' },
 ]
@@ -52,6 +52,7 @@ format.quote-style = "single"
 
 [tool.pytest.ini_options]
 addopts = "-p no:warnings"
+asyncio_mode = "auto"
 
 [project.urls]
 'Source' = 'https://github.com/mutating/locklib'

requirements_dev.txt

@@ -1,4 +1,5 @@
 pytest==8.0.2
+pytest-asyncio==0.23.8
 pytest-timeout==2.3.1
 coverage==7.6.1
 twine==6.1.0

tests/documentation/test_readme.py

@@ -5,7 +5,9 @@
 
 from locklib import (
     AsyncContextLockProtocol,
+    AsyncEmptyLock,
     ContextLockProtocol,
+    EmptyLock,
     LockProtocol,
     SmartLock,
 )
@@ -40,3 +42,23 @@ def test_almost_all_lock_are_context_locks():
 
 def test_asyncio_lock_is_async_context_lock():
     assert isinstance(ALock(), AsyncContextLockProtocol)
+
+
+def test_empty_lock_usage_and_protocols():
+    lock = EmptyLock()
+
+    with lock:
+        pass
+
+    assert isinstance(lock, LockProtocol)
+    assert isinstance(lock, ContextLockProtocol)
+
+
+async def test_async_empty_lock_usage_and_protocols():
+    lock = AsyncEmptyLock()
+
+    async with lock:
+        pass
+
+    assert isinstance(lock, LockProtocol)
+    assert isinstance(lock, AsyncContextLockProtocol)

tests/units/locks/empty/test_async_empty_lock.py

@@ -0,0 +1,89 @@
+import asyncio
+
+import pytest
+
+from locklib import AsyncEmptyLock
+
+
+async def test_acquire_does_not_raise():
+    """acquire returns None per its signature and must not raise."""
+    lock = AsyncEmptyLock()
+
+    await lock.acquire()
+
+
+def test_release_returns_none_without_prior_acquire():
+    lock = AsyncEmptyLock()
+
+    assert lock.release() is None
+
+
+async def test_double_acquire_does_not_block():
+    lock = AsyncEmptyLock()
+
+    await lock.acquire()
+    await lock.acquire()
+
+
+async def test_context_manager_binds_none():
+    async with AsyncEmptyLock() as value:
+        assert value is None
+
+
+async def test_nested_context_manager_does_not_deadlock():
+    lock = AsyncEmptyLock()
+
+    async with lock, lock:
+        pass
+
+
+async def test_exception_inside_context_manager_propagates():
+    lock = AsyncEmptyLock()
+
+    with pytest.raises(ValueError, match='kek'):
+        async with lock:
+            raise ValueError('kek')
+
+
+async def test_instance_is_reusable():
+    lock = AsyncEmptyLock()
+
+    for _ in range(3):
+        await lock.acquire()
+        lock.release()
+        async with lock:
+            pass
+
+
+async def test_no_serialization_between_coroutines():
+    """The empty lock does not serialize tasks, unlike a real lock.
+
+    Both tasks are inside the empty lock's section at once. A real ``asyncio.Lock``
+    serializes them: the second task cannot acquire it while the first holds it,
+    so the scenario deadlocks, surfaced here as a timeout so the test does not hang.
+    """
+    async def both_tasks_enter_section_together(lock):
+        """Return whether two tasks can be inside the lock's section at once.
+
+        The first task enters and waits, inside the lock, for the second one to
+        enter too. They can only meet if the lock lets both in simultaneously; a
+        real lock blocks the second task on acquire, so the scenario never ends.
+        """
+        second_entered = asyncio.Event()
+
+        async def first() -> None:
+            async with lock:
+                await second_entered.wait()
+
+        async def second() -> None:
+            async with lock:
+                second_entered.set()
+
+        await asyncio.gather(first(), second())
+
+        return second_entered.is_set()
+
+    assert await both_tasks_enter_section_together(AsyncEmptyLock()) is True
+
+    with pytest.raises(asyncio.TimeoutError):
+        await asyncio.wait_for(both_tasks_enter_section_together(asyncio.Lock()), timeout=0.1)

tests/units/locks/empty/test_empty_lock.py

@@ -0,0 +1,91 @@
+from threading import Barrier, BrokenBarrierError, Lock, Thread
+
+import pytest
+
+from locklib import EmptyLock
+
+
+def test_acquire_returns_none_and_does_not_raise():
+    lock = EmptyLock()
+
+    assert lock.acquire() is None
+
+
+def test_release_returns_none_without_prior_acquire():
+    lock = EmptyLock()
+
+    assert lock.release() is None
+
+
+def test_double_acquire_does_not_block():
+    lock = EmptyLock()
+
+    lock.acquire()
+    lock.acquire()
+
+
+def test_context_manager_binds_none():
+    with EmptyLock() as value:
+        assert value is None
+
+
+def test_nested_context_manager_does_not_deadlock():
+    lock = EmptyLock()
+
+    with lock, lock:
+        pass
+
+
+def test_exception_inside_context_manager_propagates():
+    lock = EmptyLock()
+
+    with pytest.raises(ValueError, match='kek'), lock:
+        raise ValueError('kek')
+
+
+def test_instance_is_reusable():
+    lock = EmptyLock()
+
+    for _ in range(3):
+        lock.acquire()
+        lock.release()
+        with lock:
+            pass
+
+
+def test_no_serialization_between_threads():
+    """The empty lock does not serialize threads, unlike a real lock.
+
+    Both threads pass the shared barrier together inside the empty lock, so it
+    never has to fall back on the safety timeout. A real ``threading.Lock`` lets
+    only one thread in at a time, so the second never reaches the barrier while
+    the first waits on it, and they are never inside the section together.
+    """
+    def both_threads_enter_section_together(lock, barrier_timeout):
+        """Return whether two threads can be inside the lock's section at once.
+
+        Both threads must pass a shared barrier from within the critical section,
+        so they only succeed if the lock lets them in simultaneously;
+        ``barrier_timeout`` bounds the wait so a serializing lock does not hang.
+        """
+        barrier = Barrier(2)
+        entered_together = []
+
+        def function():
+            with lock:
+                try:
+                    barrier.wait(timeout=barrier_timeout)
+                except BrokenBarrierError:
+                    return
+                entered_together.append(True)
+
+        threads = [Thread(target=function) for _ in range(2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        return len(entered_together) == 2
+
+    assert both_threads_enter_section_together(EmptyLock(), barrier_timeout=None) is True
+    assert both_threads_enter_section_together(Lock(), barrier_timeout=1) is False

tests/units/protocols/test_async_context_lock.py

@@ -7,13 +7,14 @@
 import pytest
 from full_match import match
 
-from locklib import AsyncContextLockProtocol, SmartLock
+from locklib import AsyncContextLockProtocol, AsyncEmptyLock, EmptyLock, SmartLock
 
 
 @pytest.mark.parametrize(
     'lock',  # type: ignore[no-untyped-def, unused-ignore]
     [
         ALock(),
+        AsyncEmptyLock(),
     ],
 )
 def test_locks_are_instances_of_context_lock_protocol(lock):  # type: ignore[no-untyped-def, unused-ignore]
@@ -33,6 +34,7 @@ def test_locks_are_instances_of_context_lock_protocol(lock):  # type: ignore[no-
         TLock(),
         TRLock(),
         SmartLock(),
+        EmptyLock(),
     ],
 )
 def test_other_objects_are_not_instances_of_context_lock(other):  # type: ignore[no-untyped-def, unused-ignore]
@@ -69,3 +71,4 @@ def some_function(lock: AsyncContextLockProtocol) -> AsyncContextLockProtocol:
         return lock
 
     some_function(ALock())
+    some_function(AsyncEmptyLock())