Spaces:
Sleeping
Sleeping
File size: 3,774 Bytes
1e6a9db |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
"""SQLite database helpers for document indexing schema."""
from __future__ import annotations
from pathlib import Path
import sqlite3
from typing import Iterable
PROJECT_ROOT = Path(__file__).resolve().parents[3]
DATA_DIR = PROJECT_ROOT / "data"
DEFAULT_DB_PATH = DATA_DIR / "index.db"
DDL_STATEMENTS: tuple[str, ...] = (
"""
CREATE TABLE IF NOT EXISTS note_metadata (
user_id TEXT NOT NULL,
note_path TEXT NOT NULL,
version INTEGER NOT NULL DEFAULT 1,
title TEXT NOT NULL,
created TEXT NOT NULL,
updated TEXT NOT NULL,
size_bytes INTEGER NOT NULL DEFAULT 0,
normalized_title_slug TEXT,
normalized_path_slug TEXT,
PRIMARY KEY (user_id, note_path)
)
""",
"CREATE INDEX IF NOT EXISTS idx_metadata_user ON note_metadata(user_id)",
"CREATE INDEX IF NOT EXISTS idx_metadata_updated ON note_metadata(user_id, updated DESC)",
"CREATE INDEX IF NOT EXISTS idx_metadata_title_slug ON note_metadata(user_id, normalized_title_slug)",
"CREATE INDEX IF NOT EXISTS idx_metadata_path_slug ON note_metadata(user_id, normalized_path_slug)",
"""
CREATE VIRTUAL TABLE IF NOT EXISTS note_fts USING fts5(
user_id UNINDEXED,
note_path UNINDEXED,
title,
body,
tokenize='porter unicode61',
prefix='2 3'
)
""",
"""
CREATE TABLE IF NOT EXISTS note_tags (
user_id TEXT NOT NULL,
note_path TEXT NOT NULL,
tag TEXT NOT NULL,
PRIMARY KEY (user_id, note_path, tag)
)
""",
"CREATE INDEX IF NOT EXISTS idx_tags_user_tag ON note_tags(user_id, tag)",
"CREATE INDEX IF NOT EXISTS idx_tags_user_path ON note_tags(user_id, note_path)",
"""
CREATE TABLE IF NOT EXISTS note_links (
user_id TEXT NOT NULL,
source_path TEXT NOT NULL,
target_path TEXT,
link_text TEXT NOT NULL,
is_resolved INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (user_id, source_path, link_text)
)
""",
"CREATE INDEX IF NOT EXISTS idx_links_user_source ON note_links(user_id, source_path)",
"CREATE INDEX IF NOT EXISTS idx_links_user_target ON note_links(user_id, target_path)",
"CREATE INDEX IF NOT EXISTS idx_links_unresolved ON note_links(user_id, is_resolved)",
"""
CREATE TABLE IF NOT EXISTS index_health (
user_id TEXT PRIMARY KEY,
note_count INTEGER NOT NULL DEFAULT 0,
last_full_rebuild TEXT,
last_incremental_update TEXT
)
""",
)
class DatabaseService:
"""Manage SQLite connections and schema initialization."""
def __init__(self, db_path: str | Path | None = None):
self.db_path = Path(db_path) if db_path else DEFAULT_DB_PATH
def _ensure_directory(self) -> None:
self.db_path.parent.mkdir(parents=True, exist_ok=True)
def connect(self) -> sqlite3.Connection:
"""Return a sqlite3 connection with the proper data directory created."""
self._ensure_directory()
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def initialize(self, statements: Iterable[str] | None = None) -> Path:
"""Create all schema artifacts required for indexing."""
conn = self.connect()
try:
with conn: # Transactional apply of DDL
for statement in statements or DDL_STATEMENTS:
conn.execute(statement)
finally:
conn.close()
return self.db_path
def init_database(db_path: str | Path | None = None) -> Path:
"""Convenience wrapper matching the quickstart instructions."""
return DatabaseService(db_path).initialize()
__all__ = ["DatabaseService", "init_database", "DEFAULT_DB_PATH"]
|