Coverage for session_buddy / backends / local_backend.py: 18.69%

1"""Local file system session storage backend. 

2 

3**DEPRECATED**: This module is deprecated and will be removed in v1.0. 

4Use ServerlessStorageAdapter with backend="file" instead, which uses ACB's 

5native file storage adapter with async support and better performance. 

6 

7Migration: 

8 Old: LocalFileStorage(config) 

9 New: ServerlessStorageAdapter(config, backend="file") 

10 

11This module provides a local file system implementation of the SessionStorage interface 

12for storing and retrieving session state in local files (useful for development/testing). 

13""" 

14 

15from __future__ import annotations 

16 

17import gzip 

18import json 

19import warnings 

20from datetime import datetime 

21from pathlib import Path 

22from typing import Any 

23 

24from session_buddy.backends.base import SessionState, SessionStorage 

25 

26 

27class LocalFileStorage(SessionStorage): 

28 """Local file-based session storage (for development/testing). 

29 

30 .. deprecated:: 0.9.3 

31 LocalFileStorage is deprecated. Use ``ServerlessStorageAdapter(backend="file")`` 

32 which provides async file operations and better performance via ACB. 

33 

34 """ 

35 

36 def __init__(self, config: dict[str, Any]) -> None: 

37 warnings.warn( 

38 "LocalFileStorage is deprecated and will be removed in v1.0. " 

39 "Use ServerlessStorageAdapter(backend='file') instead for " 

40 "async file operations and ACB integration.", 

41 DeprecationWarning, 

42 stacklevel=2, 

43 ) 

44 super().__init__(config) 

45 self.storage_dir = Path( 

46 config.get("storage_dir", Path.home() / ".claude" / "data" / "sessions"), 

47 ) 

48 self.storage_dir.mkdir(parents=True, exist_ok=True) 

49 

50 def _get_session_file(self, session_id: str) -> Path: 

51 """Get file path for session.""" 

52 return self.storage_dir / f"{session_id}.json.gz" 

53 

54 async def store_session( 

55 self, 

56 session_state: SessionState, 

57 ttl_seconds: int | None = None, 

58 ) -> bool: 

59 """Store session in local file.""" 

60 try: 

61 # Serialize and compress session state 

62 serialized = json.dumps(session_state.to_dict()) 

63 compressed = gzip.compress(serialized.encode("utf-8")) 

64 

65 # Write to file 

66 session_file = self._get_session_file(session_state.session_id) 

67 with session_file.open("wb") as f: 

68 f.write(compressed) 

69 

70 return True 

71 

72 except Exception as e: 

73 self.logger.exception( 

74 f"Failed to store session {session_state.session_id}: {e}", 

75 ) 

76 return False 

77 

78 async def retrieve_session(self, session_id: str) -> SessionState | None: 

79 """Retrieve session from local file.""" 

80 try: 

81 session_file = self._get_session_file(session_id) 

82 

83 if not session_file.exists(): 

84 return None 

85 

86 # Read and decompress 

87 with session_file.open("rb") as f: 

88 compressed_data = f.read() 

89 

90 serialized = gzip.decompress(compressed_data).decode("utf-8") 

91 session_data = json.loads(serialized) 

92 

93 return SessionState.from_dict(session_data) 

94 

95 except Exception as e: 

96 self.logger.exception(f"Failed to retrieve session {session_id}: {e}") 

97 return None 

98 

99 async def delete_session(self, session_id: str) -> bool: 

100 """Delete session file.""" 

101 try: 

102 session_file = self._get_session_file(session_id) 

103 

104 if session_file.exists(): 

105 session_file.unlink() 

106 return True 

107 

108 return False 

109 

110 except Exception as e: 

111 self.logger.exception(f"Failed to delete session {session_id}: {e}") 

112 return False 

113 

114 async def list_sessions( 

115 self, 

116 user_id: str | None = None, 

117 project_id: str | None = None, 

118 ) -> list[str]: 

119 """List session files.""" 

120 try: 

121 session_ids = [] 

122 for session_file in self.storage_dir.glob("*.json.gz"): 

123 session_id = self._extract_session_id(session_file) 

124 if await self._should_include_session(session_id, user_id, project_id): 

125 session_ids.append(session_id) 

126 return session_ids 

127 except Exception as e: 

128 self.logger.exception(f"Failed to list sessions: {e}") 

129 return [] 

130 

131 def _extract_session_id(self, session_file: Path) -> str: 

132 """Extract session ID from file path.""" 

133 return session_file.stem.replace(".json", "") 

134 

135 async def _should_include_session( 

136 self, 

137 session_id: str, 

138 user_id: str | None, 

139 project_id: str | None, 

140 ) -> bool: 

141 """Check if session should be included based on filters.""" 

142 if not user_id and not project_id: 

143 return True 

144 

145 session_state = await self.retrieve_session(session_id) 

146 if not session_state: 

147 return False 

148 

149 return self._matches_filters(session_state, user_id, project_id) 

150 

151 def _matches_filters( 

152 self, 

153 session_state: SessionState, 

154 user_id: str | None, 

155 project_id: str | None, 

156 ) -> bool: 

157 """Check if session matches the given filters.""" 

158 if user_id and session_state.user_id != user_id: 

159 return False 

160 return not (project_id and session_state.project_id != project_id) 

161 

162 async def cleanup_expired_sessions(self) -> int: 

163 """Clean up old session files.""" 

164 try: 

165 now = datetime.now() 

166 cleaned = 0 

167 

168 for session_file in self.storage_dir.glob("*.json.gz"): 

169 # Check file age 

170 file_age = now - datetime.fromtimestamp(session_file.stat().st_mtime) 

171 

172 if file_age.days > 7: # Cleanup sessions older than 7 days 

173 session_file.unlink() 

174 cleaned += 1 

175 

176 return cleaned 

177 

178 except Exception as e: 

179 self.logger.exception(f"Failed to cleanup expired sessions: {e}") 

180 return 0 

181 

182 async def is_available(self) -> bool: 

183 """Check if local storage is available.""" 

184 return self.storage_dir.exists() and self.storage_dir.is_dir()