Ready

backend/app/storage.py

@@ -0,0 +1,116 @@
+import json
+import sqlite3
+from collections.abc import Iterator
+from contextlib import closing, contextmanager
+from datetime import datetime, timezone
+from pathlib import Path
+
+from pydantic import BaseModel
+
+
+CREATE_NOTES = """
+    CREATE TABLE IF NOT EXISTS notes (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        title TEXT NOT NULL,
+        content TEXT NOT NULL,
+        created_at TEXT NOT NULL,
+        summary TEXT,
+        tags TEXT NOT NULL DEFAULT '[]'
+    )
+"""
+
+
+class Note(BaseModel):
+    id: int
+    title: str
+    content: str
+    created_at: str
+    summary: str | None
+    tags: list[str]
+
+
+def note_from_row(row: sqlite3.Row) -> Note:
+    fields = dict(row)
+    fields["tags"] = json.loads(fields["tags"])
+    return Note(**fields)
+
+
+class NoteStore:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+
+    @contextmanager
+    def connect(self) -> Iterator[sqlite3.Connection]:
+        with closing(sqlite3.connect(self.path)) as connection, connection:
+            connection.row_factory = sqlite3.Row
+            yield connection
+
+    def initialize(self) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self.connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            connection.execute(CREATE_NOTES)
+            schema = connection.execute(
+                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'notes'"
+            ).fetchone()["sql"]
+            if "AUTOINCREMENT" not in schema.upper():
+                # A stale tab or stream may still refer to a deleted note's ID.
+                connection.execute("ALTER TABLE notes RENAME TO notes_legacy")
+                connection.execute(CREATE_NOTES)
+                connection.execute(
+                    """
+                    INSERT INTO notes (id, title, content, created_at, summary, tags)
+                    SELECT id, title, content, created_at, summary, tags FROM notes_legacy
+                    """
+                )
+                connection.execute("DROP TABLE notes_legacy")
+
+    def list_notes(self) -> list[Note]:
+        with self.connect() as connection:
+            rows = connection.execute("SELECT * FROM notes ORDER BY id DESC")
+            return [note_from_row(row) for row in rows]
+
+    def get_note(self, note_id: int) -> Note | None:
+        with self.connect() as connection:
+            row = connection.execute("SELECT * FROM notes WHERE id = ?", (note_id,)).fetchone()
+            return note_from_row(row) if row else None
+
+    def create_note(self, title: str, content: str) -> Note:
+        created_at = datetime.now(timezone.utc).isoformat()
+        with self.connect() as connection:
+            row = connection.execute(
+                "INSERT INTO notes (title, content, created_at) VALUES (?, ?, ?) RETURNING *",
+                (title, content, created_at),
+            ).fetchone()
+            return note_from_row(row)
+
+    def delete_note(self, note_id: int) -> None:
+        with self.connect() as connection:
+            connection.execute("DELETE FROM notes WHERE id = ?", (note_id,))
+
+    def update_note(self, note_id: int, title: str, content: str) -> Note | None:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                UPDATE notes
+                SET title = :title, content = :content,
+                    summary = CASE WHEN title = :title AND content = :content THEN summary END,
+                    tags = CASE WHEN title = :title AND content = :content THEN tags ELSE '[]' END
+                WHERE id = :id
+                RETURNING *
+                """,
+                {"id": note_id, "title": title, "content": content},
+            ).fetchone()
+            return note_from_row(row) if row else None
+
+    def save_analysis(self, source: Note, summary: str | None, tags: list[str]) -> Note | None:
+        with self.connect() as connection:
+            row = connection.execute(
+                """
+                UPDATE notes SET summary = ?, tags = ?
+                WHERE id = ? AND title = ? AND content = ?
+                RETURNING *
+                """,
+                (summary, json.dumps(tags, ensure_ascii=False), source.id, source.title, source.content),
+            ).fetchone()
+            return note_from_row(row) if row else None