diff options
| author | batsumaru <> | 2026-09-15 15:13:34 +0900 |
|---|---|---|
| committer | batsumaru <> | 2026-09-15 15:13:34 +0900 |
| commit | 1a67c5edb6fa6165e99cc25ed380303ed817af09 (patch) | |
| tree | d2eab77e7275a559bfd639fd80b15321b5e514f3 /memory_store.py | |
| parent | 5629cabbb38c7d91e66a419ba7cb3a93a9a988a1 (diff) | |
Cleaned up project tree
Diffstat (limited to 'memory_store.py')
| -rw-r--r-- | memory_store.py | 99 |
1 files changed, 0 insertions, 99 deletions
diff --git a/memory_store.py b/memory_store.py deleted file mode 100644 index 2c69113..0000000 --- a/memory_store.py +++ /dev/null @@ -1,99 +0,0 @@ -from datetime import datetime -from uuid import UUID - -from database.database import Database -from memory import LongTermMemory, MemoryCategory - - -class LongTermMemoryStore: - def __init__(self, database: Database) -> None: - self.database = database - - def add(self, memory: LongTermMemory) -> None: - created_at = memory.created_at.isoformat().replace("+00:00", "Z") - - with self.database.connect() as connection: - connection.execute( - """ - INSERT INTO memories (id, content, category, created_at) - VALUES (?, ?, ?, ?) - """, - (str(memory.id), memory.content, memory.category.value, created_at), - ) - - def get(self, memory_id: UUID) -> LongTermMemory | None: - with self.database.connect() as connection: - row = connection.execute( - """ - SELECT id, content, category, created_at - FROM memories - WHERE id = ? - """, - (str(memory_id),), - ).fetchone() - - if row is None: - return None - - return self._from_row(row) - - def list(self) -> list[LongTermMemory]: - with self.database.connect() as connection: - rows = connection.execute( - """ - SELECT id, content, category, created_at - FROM memories - ORDER BY id - """ - ).fetchall() - - return [self._from_row(row) for row in rows] - - def delete(self, memory_id: UUID) -> None: - with self.database.connect() as connection: - connection.execute( - "DELETE FROM memories WHERE id = ?", - (str(memory_id),), - ) - - def search( - self, - query: str, - category: MemoryCategory | None = None, - limit: int = 10, - ) -> list[LongTermMemory]: - if limit <= 0: - raise ValueError("Search limit must be positive") - - if not query.strip(): - raise ValueError("Search query must not be empty") - - sql = """ - SELECT id, content, category, created_at - FROM memories - WHERE content LIKE ? - """ - parameters: list[str | int] = [f"%{query}%"] - - if category is not None: - sql += " AND category = ?" - parameters.append(category.value) - - sql += " ORDER BY id LIMIT ?" - parameters.append(limit) - - with self.database.connect() as connection: - rows = connection.execute(sql, parameters).fetchall() - - return [self._from_row(row) for row in rows] - - @staticmethod - def _from_row(row: tuple[str, str, str, str]) -> LongTermMemory: - memory_id, content, category, created_at = row - - return LongTermMemory( - id=UUID(memory_id), - content=content, - category=MemoryCategory(category), - created_at=datetime.fromisoformat(created_at), - ) |
