kodi
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
class BookmarkRepository:
|
|
def __init__(self, db_path: str):
|
|
self._db_path = db_path
|
|
self._ensure_schema()
|
|
|
|
def create_bookmark(self, path: str, label: str) -> dict:
|
|
created_at = self._now_iso()
|
|
with self._connection() as conn:
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO bookmarks (path, label, created_at)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(path, label, created_at),
|
|
)
|
|
bookmark_id = int(cursor.lastrowid)
|
|
row = conn.execute(
|
|
"SELECT id, path, label, created_at FROM bookmarks WHERE id = ?",
|
|
(bookmark_id,),
|
|
).fetchone()
|
|
return self._to_dict(row)
|
|
|
|
def list_bookmarks(self) -> list[dict]:
|
|
with self._connection() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, path, label, created_at
|
|
FROM bookmarks
|
|
ORDER BY created_at DESC
|
|
"""
|
|
).fetchall()
|
|
return [self._to_dict(row) for row in rows]
|
|
|
|
def delete_bookmark(self, bookmark_id: int) -> bool:
|
|
with self._connection() as conn:
|
|
cursor = conn.execute("DELETE FROM bookmarks WHERE id = ?", (bookmark_id,))
|
|
return cursor.rowcount > 0
|
|
|
|
def _ensure_schema(self) -> None:
|
|
db_path = Path(self._db_path)
|
|
if db_path.parent and str(db_path.parent) not in {"", "."}:
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with self._connection() as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS bookmarks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
path TEXT NOT NULL UNIQUE,
|
|
label TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_bookmarks_created_at_desc
|
|
ON bookmarks(created_at DESC)
|
|
"""
|
|
)
|
|
|
|
@contextmanager
|
|
def _connection(self):
|
|
conn = sqlite3.connect(self._db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
@staticmethod
|
|
def _to_dict(row: sqlite3.Row) -> dict:
|
|
return {
|
|
"id": int(row["id"]),
|
|
"path": row["path"],
|
|
"label": row["label"],
|
|
"created_at": row["created_at"],
|
|
}
|
|
|
|
@staticmethod
|
|
def _now_iso() -> str:
|
|
return datetime.now(tz=timezone.utc).isoformat().replace("+00:00", "Z")
|