| """SQLite database layer for Prompt Garage.""" |
|
|
| import os |
| import sqlite3 |
|
|
| DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "prompts.db") |
|
|
| SCHEMA = """ |
| CREATE TABLE IF NOT EXISTS prompts ( |
| id INTEGER PRIMARY KEY, |
| title TEXT NOT NULL, |
| content TEXT NOT NULL, |
| source TEXT, |
| is_starred INTEGER DEFAULT 0, |
| tags TEXT, |
| model TEXT, |
| rating INTEGER, |
| use_count INTEGER DEFAULT 0, |
| created_at TEXT DEFAULT (datetime('now')), |
| updated_at TEXT DEFAULT (datetime('now')) |
| ); |
| |
| CREATE VIRTUAL TABLE IF NOT EXISTS prompts_fts USING fts5( |
| title, content, tags, content=prompts, content_rowid=id |
| ); |
| |
| CREATE TRIGGER IF NOT EXISTS prompts_ai AFTER INSERT ON prompts BEGIN |
| INSERT INTO prompts_fts(rowid, title, content, tags) |
| VALUES (new.id, new.title, new.content, new.tags); |
| END; |
| |
| CREATE TRIGGER IF NOT EXISTS prompts_ad AFTER DELETE ON prompts BEGIN |
| INSERT INTO prompts_fts(prompts_fts, rowid, title, content, tags) |
| VALUES('delete', old.id, old.title, old.content, old.tags); |
| END; |
| |
| CREATE TRIGGER IF NOT EXISTS prompts_au AFTER UPDATE ON prompts BEGIN |
| INSERT INTO prompts_fts(prompts_fts, rowid, title, content, tags) |
| VALUES('delete', old.id, old.title, old.content, old.tags); |
| INSERT INTO prompts_fts(rowid, title, content, tags) |
| VALUES (new.id, new.title, new.content, new.tags); |
| END; |
| """ |
|
|
|
|
| def get_db() -> sqlite3.Connection: |
| conn = sqlite3.connect(DB_PATH) |
| conn.row_factory = sqlite3.Row |
| conn.execute("PRAGMA journal_mode=WAL") |
| return conn |
|
|
|
|
| def init_db(): |
| conn = get_db() |
| conn.executescript(SCHEMA) |
| conn.close() |
|
|
|
|
| def add_prompt(title, content, source="manual", tags=None, model=None, rating=None, is_starred=0): |
| conn = get_db() |
| cur = conn.execute( |
| "INSERT INTO prompts (title, content, source, tags, model, rating, is_starred) VALUES (?, ?, ?, ?, ?, ?, ?)", |
| (title, content, source, tags, model, rating, is_starred), |
| ) |
| conn.commit() |
| pid = cur.lastrowid |
| conn.close() |
| return pid |
|
|
|
|
| def list_prompts(starred=None, tag=None, model=None, limit=20): |
| conn = get_db() |
| query = "SELECT * FROM prompts WHERE 1=1" |
| params = [] |
| if starred: |
| query += " AND is_starred = 1" |
| if tag: |
| query += " AND tags LIKE ?" |
| params.append(f"%{tag}%") |
| if model: |
| query += " AND model = ?" |
| params.append(model) |
| query += " ORDER BY updated_at DESC LIMIT ?" |
| params.append(limit) |
| rows = conn.execute(query, params).fetchall() |
| conn.close() |
| return rows |
|
|
|
|
| def get_prompt(pid): |
| conn = get_db() |
| row = conn.execute("SELECT * FROM prompts WHERE id = ?", (pid,)).fetchone() |
| conn.close() |
| return row |
|
|
|
|
| def delete_prompt(pid): |
| conn = get_db() |
| conn.execute("DELETE FROM prompts WHERE id = ?", (pid,)) |
| conn.commit() |
| conn.close() |
|
|
|
|
| def update_prompt(pid, **kwargs): |
| conn = get_db() |
| sets = ", ".join(f"{k} = ?" for k in kwargs) |
| vals = list(kwargs.values()) + [pid] |
| conn.execute(f"UPDATE prompts SET {sets}, updated_at = datetime('now') WHERE id = ?", vals) |
| conn.commit() |
| conn.close() |
|
|
|
|
| def toggle_star(pid): |
| conn = get_db() |
| conn.execute("UPDATE prompts SET is_starred = 1 - is_starred, updated_at = datetime('now') WHERE id = ?", (pid,)) |
| conn.commit() |
| row = conn.execute("SELECT is_starred FROM prompts WHERE id = ?", (pid,)).fetchone() |
| conn.close() |
| return row["is_starred"] if row else None |
|
|
|
|
| def set_rating(pid, rating): |
| update_prompt(pid, rating=rating) |
|
|
|
|
| def increment_use(pid): |
| conn = get_db() |
| conn.execute("UPDATE prompts SET use_count = use_count + 1, updated_at = datetime('now') WHERE id = ?", (pid,)) |
| conn.commit() |
| conn.close() |
|
|
|
|
| def search_fts(query): |
| conn = get_db() |
| rows = conn.execute( |
| "SELECT p.* FROM prompts p JOIN prompts_fts f ON p.id = f.rowid WHERE prompts_fts MATCH ? ORDER BY rank", |
| (query,), |
| ).fetchall() |
| conn.close() |
| return rows |
|
|
|
|
| def get_stats(): |
| conn = get_db() |
| total = conn.execute("SELECT COUNT(*) as c FROM prompts").fetchone()["c"] |
| starred = conn.execute("SELECT COUNT(*) as c FROM prompts WHERE is_starred = 1").fetchone()["c"] |
| by_source = conn.execute("SELECT source, COUNT(*) as c FROM prompts GROUP BY source").fetchall() |
| by_model = conn.execute("SELECT model, COUNT(*) as c FROM prompts WHERE model IS NOT NULL GROUP BY model").fetchall() |
| by_tag_raw = conn.execute("SELECT tags FROM prompts WHERE tags IS NOT NULL AND tags != ''").fetchall() |
| conn.close() |
|
|
| tag_counts = {} |
| for row in by_tag_raw: |
| for t in row["tags"].split(","): |
| t = t.strip() |
| if t: |
| tag_counts[t] = tag_counts.get(t, 0) + 1 |
|
|
| return { |
| "total": total, |
| "starred": starred, |
| "by_source": {r["source"]: r["c"] for r in by_source}, |
| "by_model": {r["model"]: r["c"] for r in by_model}, |
| "by_tag": tag_counts, |
| } |
|
|
|
|
| def list_top_used(limit=20): |
| """Return prompts with use_count > 0, sorted by use_count descending.""" |
| conn = get_db() |
| rows = conn.execute( |
| "SELECT * FROM prompts WHERE use_count > 0 ORDER BY use_count DESC LIMIT ?", |
| (limit,), |
| ).fetchall() |
| conn.close() |
| return rows |
|
|
|
|
| def list_untagged(limit=100): |
| """Return prompts that have no tags (NULL or empty string).""" |
| conn = get_db() |
| rows = conn.execute( |
| "SELECT * FROM prompts WHERE tags IS NULL OR tags = '' LIMIT ?", |
| (limit,), |
| ).fetchall() |
| conn.close() |
| return rows |
|
|
|
|
| def prompt_exists_by_content(content_hash): |
| """Check if a prompt with this content already exists (dedup by hash stored in source field).""" |
| conn = get_db() |
| row = conn.execute("SELECT id FROM prompts WHERE source LIKE ?", (f"%{content_hash}%",)).fetchone() |
| conn.close() |
| return row is not None |
|
|
