117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
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
|