Coverage for src / agent_contracts / store / memory.py: 0%

1"""InMemoryStateStore - In-memory state storage. 

2 

3Simple in-process storage for development and testing. 

4Not suitable for production use with multiple processes. 

5""" 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from agent_contracts.store.base import StateStore 

11 

12 

13class InMemoryStateStore(StateStore): 

14 """In-memory state storage. 

15  

16 Simple in-process dictionary storage. 

17 Useful for development and testing. 

18  

19 Example: 

20 store = InMemoryStateStore() 

21 await store.save("user:123", {"name": "John"}) 

22 data = await store.load("user:123") 

23 """ 

24 

25 def __init__(self) -> None: 

26 self._data: dict[str, dict[str, Any]] = {} 

27 

28 async def save(self, key: str, value: dict[str, Any]) -> None: 

29 self._data[key] = value 

30 

31 async def load(self, key: str) -> dict[str, Any] | None: 

32 return self._data.get(key) 

33 

34 async def delete(self, key: str) -> bool: 

35 if key in self._data: 

36 del self._data[key] 

37 return True 

38 return False 

39 

40 async def exists(self, key: str) -> bool: 

41 return key in self._data 

42 

43 async def list_by_prefix(self, prefix: str) -> list[str]: 

44 return [k for k in self._data.keys() if k.startswith(prefix)] 

45 

46 def clear(self) -> None: 

47 """Clear all stored data (for testing).""" 

48 self._data.clear()