Added long-term memory

1 files changed, 99 insertions, 0 deletions

diff --git a/memory_store.py b/memory_store.py
new file mode 100644
index 0000000..2c69113
--- /dev/null
+++ b/memory_store.py
@@ -0,0 +1,99 @@
+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),
+        )