Remove sessions with no messages
(self)
| 241 | return deleted |
| 242 | |
| 243 | def cleanup_empty_sessions(self) -> int: |
| 244 | """Remove sessions with no messages""" |
| 245 | conn = sqlite3.connect(self.db_path) |
| 246 | |
| 247 | # Find sessions with no messages |
| 248 | cursor = conn.execute(''' |
| 249 | SELECT s.id FROM sessions s |
| 250 | LEFT JOIN messages m ON s.id = m.session_id |
| 251 | WHERE m.id IS NULL |
| 252 | ''') |
| 253 | |
| 254 | empty_sessions = [row[0] for row in cursor.fetchall()] |
| 255 | |
| 256 | # Delete empty sessions |
| 257 | deleted_count = 0 |
| 258 | for session_id in empty_sessions: |
| 259 | cursor = conn.execute('DELETE FROM sessions WHERE id = ?', (session_id,)) |
| 260 | if cursor.rowcount > 0: |
| 261 | deleted_count += 1 |
| 262 | print(f"🗑️ Cleaned up empty session: {session_id[:8]}...") |
| 263 | |
| 264 | conn.commit() |
| 265 | conn.close() |
| 266 | |
| 267 | if deleted_count > 0: |
| 268 | print(f"✨ Cleaned up {deleted_count} empty sessions") |
| 269 | |
| 270 | return deleted_count |
| 271 | |
| 272 | def get_stats(self) -> Dict: |
| 273 | """Get database statistics""" |
no outgoing calls
no test coverage detected