Coverage for src / lexigram / contracts / core / stores.py: 0%

1"""Persistent distributed lock store protocol. 

2 

3Provides advisory, TTL-based distributed locks backed by a persistent store 

4(SQL advisory locks, a SQL ``locks`` table, Redis, etc.), allowing multiple 

5services to coordinate exclusive access to shared resources. 

6 

7Implementations ship in ``lexigram-sql`` (``lexigram.sql.stores``) so they can 

8share the application's database connection pool without creating a lateral 

9cross-extension dependency. 

10 

11Example:: 

12 

13 from lexigram.contracts.core.stores import LockStoreProtocol 

14 

15 store = await container.resolve(LockStoreProtocol) 

16 acquired = await store.acquire("billing_job", owner=instance_id, ttl=60) 

17 if acquired: 

18 try: 

19 ... 

20 finally: 

21 await store.release("billing_job", owner=instance_id) 

22""" 

23 

24from __future__ import annotations 

25 

26from typing import Protocol, runtime_checkable 

27 

28 

29@runtime_checkable 

30class LockStoreProtocol(Protocol): 

31 """Persistent distributed lock store. 

32 

33 Individual lock handles are *not* returned; all operations are done via 

34 the lock name and owner identifier. 

35 """ 

36 

37 async def acquire(self, lock_name: str, owner: str, ttl: int) -> bool: 

38 """Attempt to acquire the named lock. 

39 

40 Args: 

41 lock_name: Globally unique lock identifier. 

42 owner: Identifier of the requesting owner (e.g. host + PID). 

43 ttl: Lock time-to-live in seconds. The lock is automatically 

44 released after this duration to prevent deadlocks. 

45 

46 Returns: 

47 ``True`` if the lock was acquired; ``False`` if it is already held. 

48 """ 

49 ... 

50 

51 async def release(self, lock_name: str, owner: str) -> bool: 

52 """Release the named lock *only if* it is held by *owner*. 

53 

54 Args: 

55 lock_name: Globally unique lock identifier. 

56 owner: Must match the owner that acquired the lock. 

57 

58 Returns: 

59 ``True`` if the lock was released; ``False`` if it was not held 

60 by *owner* or did not exist. 

61 """ 

62 ... 

63 

64 async def extend(self, lock_name: str, owner: str, ttl: int) -> bool: 

65 """Extend the TTL of a currently-held lock. 

66 

67 Args: 

68 lock_name: Globally unique lock identifier. 

69 owner: Must match the owner that currently holds the lock. 

70 ttl: New time-to-live in seconds from now. 

71 

72 Returns: 

73 ``True`` if the extension was applied; ``False`` if the lock was 

74 not held by *owner*. 

75 """ 

76 ... 

77 

78 

79__all__ = ["LockStoreProtocol"]