Coverage for src / lexigram / contracts / core / lock.py: 0%
51 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-15 18:57 +0800
1"""Distributed Lock protocol for Lexigram Framework.
3Provides the DistributedLockProtocol protocol for distributed locking across
4multiple processes or services. Implementations may use Redis,
5PostgreSQL advisory locks, ZooKeeper, or other coordination services.
7Example::
9 from lexigram.contracts.lock import DistributedLockProtocol
11 async def process_with_lock(lock: DistributedLockProtocol) -> None:
12 if await lock.acquire():
13 try:
14 # Critical section - only one process holds the lock
15 await process_data()
16 finally:
17 await lock.release()
18 else:
19 logger.info("Could not acquire lock, skipping")
21 # Or use as context manager
22 async with lock:
23 await process_data()
24"""
26from __future__ import annotations
28from abc import abstractmethod
29from dataclasses import dataclass
30from typing import TYPE_CHECKING, Any, Protocol, Self, runtime_checkable
32if TYPE_CHECKING:
33 from contextlib import AbstractAsyncContextManager
34 from types import TracebackType
37@runtime_checkable
38class DistributedLockProtocol(Protocol):
39 """Protocol for distributed locking.
41 Distributed locks coordinate access to shared resources across
42 multiple processes, threads, or services. They provide:
44 - Mutual exclusion: Only one holder at a time
45 - Safety: Locks expire after a TTL to prevent deadlocks
46 - Non-blocking acquire: Returns immediately if lock is held
48 Implementations must be safe for use in async contexts and
49 handle network partitions gracefully.
51 Example::
53 lock = RedisLock(redis, "my-resource", ttl=30)
55 # Non-blocking acquire
56 if await lock.acquire():
57 try:
58 await process()
59 finally:
60 await lock.release()
62 # Blocking with timeout
63 if await lock.acquire_blocking(timeout=5.0):
64 try:
65 await process()
66 finally:
67 await lock.release()
69 # Context manager (preferred)
70 async with lock:
71 await process()
72 """
74 @abstractmethod
75 async def acquire(self) -> bool:
76 """Attempt to acquire the lock non-blocking.
78 Returns immediately with True if the lock was acquired,
79 False if the lock is already held by another process.
81 Returns:
82 True if lock was acquired, False otherwise
83 """
84 ...
86 @abstractmethod
87 async def acquire_blocking(self, timeout: float | None = None) -> bool:
88 """Attempt to acquire the lock with optional timeout.
90 Blocks until the lock is acquired or the timeout expires.
92 Args:
93 timeout: Maximum seconds to wait. None means wait forever.
95 Returns:
96 True if lock was acquired, False if timeout expired
97 """
98 ...
100 @abstractmethod
101 async def release(self) -> bool:
102 """Release the lock.
104 Returns:
105 True if lock was released by this call,
106 False if lock was not held or already expired
107 """
108 ...
110 @abstractmethod
111 async def is_held(self) -> bool:
112 """Check if this instance currently holds the lock.
114 Returns:
115 True if this instance holds the lock
116 """
117 ...
119 @abstractmethod
120 async def extend(self, additional_time: float) -> bool:
121 """Extend the lock TTL.
123 Useful for long-running operations that need to
124 prevent the lock from expiring mid-operation.
126 Args:
127 additional_time: Seconds to add to the TTL
129 Returns:
130 True if TTL was extended, False if lock not held
131 """
132 ...
134 @abstractmethod
135 async def __aenter__(self) -> DistributedLockProtocol:
136 """Enter context manager, acquiring the lock.
138 Raises:
139 LockAcquisitionError: If lock cannot be acquired
140 """
141 ...
143 @abstractmethod
144 async def __aexit__(
145 self,
146 exc_type: type[BaseException] | None,
147 exc_val: BaseException | None,
148 exc_tb: TracebackType | None,
149 ) -> None:
150 """Exit context manager, releasing the lock."""
151 ...
154@dataclass(frozen=True)
155class LockInfo:
156 """Immutable snapshot of a distributed lock's state.
158 Attributes:
159 resource: The resource being locked.
160 holder: Identifier of the current holder.
161 acquired_at: Unix timestamp when the lock was acquired.
162 expires_at: Unix timestamp when the lock expires (TTL).
163 metadata: Additional implementation-specific data.
164 """
166 resource: str
167 holder: str | None = None
168 acquired_at: float | None = None
169 expires_at: float | None = None
170 metadata: dict[str, Any] | None = None
172 @property
173 def is_expired(self) -> bool:
174 """Check if the lock has expired."""
175 from time import time
177 if self.expires_at is None:
178 return False
179 return time() > self.expires_at
181 @property
182 def ttl_remaining(self) -> float | None:
183 """Get remaining TTL in seconds."""
184 from time import time
186 if self.expires_at is None:
187 return None
188 return max(0.0, self.expires_at - time())
191@runtime_checkable
192class AsyncLockProtocol(Protocol):
193 """Protocol for a simple in-process async mutual-exclusion lock.
195 Mirrors the interface of :class:`asyncio.Lock`. Implementations are
196 **single-process** (i.e. not distributed); use :class:`DistributedLockProtocol`
197 or :class:`LockManagerProtocol` when cross-process coordination is
198 required.
200 Example::
202 from lexigram.contracts.lock import AsyncLockProtocol
204 async def protected(lock: AsyncLockProtocol) -> None:
205 async with lock:
206 await do_critical_work()
208 Container registration::
210 container.singleton(AsyncLockProtocol, InMemoryAsyncLock)
211 """
213 async def acquire(self) -> bool:
214 """Acquire the lock, blocking until it becomes available.
216 Returns:
217 ``True`` once the lock has been acquired.
218 """
219 ...
221 def release(self) -> None:
222 """Release the lock.
224 Raises:
225 RuntimeError: If the lock is not currently held.
226 """
227 ...
229 def locked(self) -> bool:
230 """Return ``True`` if the lock is currently held by any coroutine."""
231 ...
233 async def __aenter__(self) -> Self:
234 """Acquire the lock on context entry."""
235 ...
237 async def __aexit__(
238 self,
239 exc_type: type[BaseException] | None,
240 exc_val: BaseException | None,
241 exc_tb: TracebackType | None,
242 ) -> None:
243 """Release the lock on context exit."""
244 ...
247@runtime_checkable
248class LockManagerProtocol(Protocol):
249 """Protocol for a process-local or distributed lock manager.
251 A lock manager creates and tracks named locks for coordinating mutual
252 exclusion of concurrent operations. Callers use :meth:`acquire` to
253 obtain a per-key async context manager.
255 Implementations range from purely in-memory (single-process deduplication)
256 to Redis-backed or advisory-lock-backed distributed variants.
258 Example::
260 from lexigram.contracts.lock import LockManagerProtocol
262 async def process(manager: LockManagerProtocol) -> None:
263 async with manager.acquire("resource:42", timeout=30):
264 await do_work()
266 Container registration::
268 container.singleton(LockManagerProtocol, InMemoryLockManager)
269 """
271 @abstractmethod
272 def acquire(
273 self, key: str, timeout: float = 60.0
274 ) -> AbstractAsyncContextManager[Any]:
275 """Return an async context manager that holds the named lock.
277 The caller **must** use the returned value as an async context manager.
278 The lock is acquired on ``__aenter__`` and released on ``__aexit__``.
280 Args:
281 key: Unique, stable identifier for the resource to protect.
282 timeout: Informational TTL in seconds; implementations may use
283 this to automatically expire stale locks.
285 Returns:
286 An ``AbstractAsyncContextManager`` for the named lock.
287 """
288 ...
291__all__ = [
292 "AsyncLockProtocol",
293 "DistributedLockProtocol",
294 "LockInfo",
295 "LockManagerProtocol",
296]