Coverage for src / invariant / store / memory.py: 96.67%

1"""MemoryStore: In-memory artifact storage for testing.""" 

2 

3from typing import Any 

4 

5from invariant.cacheable import is_cacheable 

6from invariant.store.base import ArtifactStore 

7 

8 

9class MemoryStore(ArtifactStore): 

10 """In-memory artifact store using a dictionary. 

11 

12 Fast and ephemeral - suitable for testing. Artifacts are lost when 

13 the store instance is destroyed. 

14 """ 

15 

16 def __init__(self) -> None: 

17 """Initialize an empty memory store.""" 

18 super().__init__() 

19 self._artifacts: dict[str, Any] = {} 

20 

21 def _make_key(self, op_name: str, digest: str) -> str: 

22 """Create a composite key from op_name and digest.""" 

23 return f"{op_name}:{digest}" 

24 

25 def exists(self, op_name: str, digest: str) -> bool: 

26 """Check if an artifact exists.""" 

27 key = self._make_key(op_name, digest) 

28 exists = key in self._artifacts 

29 if exists: 

30 self.stats.hits += 1 

31 else: 

32 self.stats.misses += 1 

33 return exists 

34 

35 def get(self, op_name: str, digest: str) -> Any: 

36 """Retrieve an artifact by operation name and digest. 

37 

38 Raises: 

39 KeyError: If artifact does not exist. 

40 """ 

41 key = self._make_key(op_name, digest) 

42 if key not in self._artifacts: 

43 raise KeyError( 

44 f"Artifact with op_name '{op_name}' and digest '{digest}' not found" 

45 ) 

46 

47 # Return stored object directly (no deserialization needed) 

48 return self._artifacts[key] 

49 

50 def put(self, op_name: str, digest: str, artifact: Any) -> None: 

51 """Store an artifact with the given operation name and digest.""" 

52 # Validate artifact is cacheable 

53 if not is_cacheable(artifact): 

54 raise TypeError( 

55 f"Artifact is not cacheable: {type(artifact)}. " 

56 f"Use is_cacheable() to check values before storing." 

57 ) 

58 

59 # Store object directly (no serialization needed - relies on immutability contract) 

60 key = self._make_key(op_name, digest) 

61 self._artifacts[key] = artifact 

62 self.stats.puts += 1 

63 

64 def clear(self) -> None: 

65 """Clear all artifacts (mainly for testing).""" 

66 self._artifacts.clear() 

67 self.reset_stats()