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

1"""Idempotency store contract.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.core.result import Result 

9 from lexigram.contracts.domain.idempotency import IdempotencyRecord 

10 from lexigram.contracts.exceptions.idempotency import IdempotencyError 

11 

12 

13@runtime_checkable 

14class IdempotencyStoreProtocol(Protocol): 

15 """Protocol for storing and checking idempotency keys.""" 

16 

17 async def get(self, key: str) -> Result[Any | None, IdempotencyError]: 

18 """Retrieve a stored result by idempotency key. 

19 

20 Args: 

21 key: The idempotency key. 

22 

23 Returns: 

24 ``Ok(value)`` when a cached result exists. 

25 ``Ok(None)`` when the key is not present or expired. 

26 ``Err(IdempotencyError)`` on store failures. 

27 """ 

28 ... 

29 

30 async def get_record( 

31 self, 

32 key: str, 

33 ) -> Result[IdempotencyRecord | None, IdempotencyError]: 

34 """Retrieve the full idempotency record, including metadata.""" 

35 ... 

36 

37 async def set( 

38 self, 

39 key: str, 

40 value: Any, 

41 ttl: float | None = None, 

42 ) -> Result[None, IdempotencyError]: 

43 """Store a result with an optional time-to-live. 

44 

45 Args: 

46 key: The idempotency key. 

47 value: The result to store. 

48 ttl: Time-to-live in seconds, or ``None`` for no expiry. 

49 """ 

50 ... 

51 

52 async def delete(self, key: str) -> Result[None, IdempotencyError]: 

53 """Remove an idempotency record by key.""" 

54 ... 

55 

56 async def acquire(self, key: str, ttl: int) -> Result[bool, IdempotencyError]: 

57 """Atomically claim an idempotency key if it is not already held. 

58 

59 Args: 

60 key: The idempotency key to acquire. 

61 ttl: Time-to-live in seconds for the claimed key. 

62 

63 Returns: 

64 ``Ok(True)`` if this caller should proceed. 

65 ``Ok(False)`` if the key is already claimed. 

66 ``Err(IdempotencyError)`` on store failures. 

67 """ 

68 ... 

69 

70 

71@runtime_checkable 

72class IdempotencyMiddlewareProtocol(Protocol): 

73 """Protocol for HTTP idempotency deduplication middleware.""" 

74 

75 async def process( 

76 self, 

77 headers: dict[str, str], 

78 handler: Any, 

79 *args: Any, 

80 **kwargs: Any, 

81 ) -> Any: 

82 """Run the handler with idempotency deduplication.""" 

83 ... 

84 

85 @property 

86 def ttl(self) -> float: 

87 """Default TTL (seconds) for cached idempotency results.""" 

88 ... 

89 

90 

91__all__ = [ 

92 "IdempotencyMiddlewareProtocol", 

93 "IdempotencyStoreProtocol", 

94]