summaryrefslogtreecommitdiff
path: root/memory_store.py
diff options
context:
space:
mode:
Diffstat (limited to 'memory_store.py')
-rw-r--r--memory_store.py99
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),
- )