From 387d2465b0135f6839081e913b4132dd54055d4d Mon Sep 17 00:00:00 2001 From: gahusb Date: Mon, 25 May 2026 18:23:44 +0900 Subject: [PATCH] =?UTF-8?q?feat(tarot-lab):=20db.py=20CRUD=205=20+=20init?= =?UTF-8?q?=5Fdb=20(=ED=85=8C=EC=8A=A4=ED=8A=B8=204=EA=B1=B4=20=ED=86=B5?= =?UTF-8?q?=EA=B3=BC)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tarot-lab/app/db.py | 159 +++++++++++++++++++++++++++++++++++++ tarot-lab/tests/test_db.py | 70 ++++++++++++++++ 2 files changed, 229 insertions(+) create mode 100644 tarot-lab/app/db.py create mode 100644 tarot-lab/tests/test_db.py diff --git a/tarot-lab/app/db.py b/tarot-lab/app/db.py new file mode 100644 index 0000000..dd838ed --- /dev/null +++ b/tarot-lab/app/db.py @@ -0,0 +1,159 @@ +"""tarot.db SQLite — 5 CRUD + _tarot_row_to_dict + init_db.""" +import json +import os +import sqlite3 +from typing import Any, Dict, Optional + +from .config import DB_PATH + + +def _conn() -> sqlite3.Connection: + os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) + conn = sqlite3.connect(DB_PATH, timeout=120.0) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=120000") + return conn + + +def init_db() -> None: + with _conn() as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS tarot_readings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + spread_type TEXT NOT NULL, + category TEXT, + question TEXT, + cards TEXT NOT NULL, + interpretation_json TEXT, + summary TEXT, + model TEXT, + tokens_in INTEGER, + tokens_out INTEGER, + cost_usd REAL, + confidence TEXT, + favorite INTEGER NOT NULL DEFAULT 0, + note TEXT + ) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_tarot_created + ON tarot_readings(created_at DESC) + """) + conn.execute(""" + CREATE INDEX IF NOT EXISTS idx_tarot_favorite + ON tarot_readings(favorite, created_at DESC) + """) + + +def save_tarot_reading(data: Dict[str, Any]) -> int: + interp = data.get("interpretation_json") or {} + summary = interp.get("summary", "") if isinstance(interp, dict) else "" + with _conn() as conn: + cur = conn.execute( + """INSERT INTO tarot_readings + (spread_type, category, question, cards, interpretation_json, + summary, model, tokens_in, tokens_out, cost_usd, confidence) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + ( + data["spread_type"], + data.get("category"), + data.get("question"), + json.dumps(data.get("cards") or [], ensure_ascii=False), + json.dumps(interp, ensure_ascii=False) if interp else None, + summary, + data.get("model"), + data.get("tokens_in"), + data.get("tokens_out"), + data.get("cost_usd"), + data.get("confidence"), + ), + ) + return int(cur.lastrowid) + + +def get_tarot_reading(reading_id: int) -> Optional[Dict[str, Any]]: + with _conn() as conn: + r = conn.execute("SELECT * FROM tarot_readings WHERE id=?", (reading_id,)).fetchone() + return _tarot_row_to_dict(r) if r else None + + +def list_tarot_readings( + page: int = 1, size: int = 20, + favorite: Optional[bool] = None, + spread_type: Optional[str] = None, + category: Optional[str] = None, +) -> Dict[str, Any]: + wheres, params = [], [] + if favorite is not None: + wheres.append("favorite=?") + params.append(1 if favorite else 0) + if spread_type: + wheres.append("spread_type=?") + params.append(spread_type) + if category: + wheres.append("category=?") + params.append(category) + where_sql = ("WHERE " + " AND ".join(wheres)) if wheres else "" + offset = (page - 1) * size + with _conn() as conn: + total = conn.execute( + f"SELECT COUNT(*) c FROM tarot_readings {where_sql}", params + ).fetchone()["c"] + rows = conn.execute( + f"SELECT * FROM tarot_readings {where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?", + params + [size, offset], + ).fetchall() + return { + "items": [_tarot_row_to_dict(r) for r in rows], + "page": page, "size": size, "total": int(total), + } + + +def update_tarot_reading(reading_id: int, **kwargs) -> None: + sets, vals = [], [] + if "favorite" in kwargs and kwargs["favorite"] is not None: + sets.append("favorite=?") + vals.append(1 if kwargs["favorite"] else 0) + if "note" in kwargs and kwargs["note"] is not None: + sets.append("note=?") + vals.append(kwargs["note"]) + if not sets: + return + vals.append(reading_id) + with _conn() as conn: + conn.execute(f"UPDATE tarot_readings SET {','.join(sets)} WHERE id=?", vals) + + +def delete_tarot_reading(reading_id: int) -> None: + with _conn() as conn: + conn.execute("DELETE FROM tarot_readings WHERE id=?", (reading_id,)) + + +def _tarot_row_to_dict(r) -> Dict[str, Any]: + try: + interp = json.loads(r["interpretation_json"]) if r["interpretation_json"] else None + except (ValueError, TypeError): + interp = None + try: + cards = json.loads(r["cards"]) if r["cards"] else [] + except (ValueError, TypeError): + cards = [] + return { + "id": r["id"], + "created_at": r["created_at"], + "spread_type": r["spread_type"], + "category": r["category"], + "question": r["question"], + "cards": cards, + "interpretation_json": interp, + "summary": r["summary"], + "model": r["model"], + "tokens_in": r["tokens_in"], + "tokens_out": r["tokens_out"], + "cost_usd": r["cost_usd"], + "confidence": r["confidence"], + "favorite": int(r["favorite"]), + "note": r["note"], + } diff --git a/tarot-lab/tests/test_db.py b/tarot-lab/tests/test_db.py new file mode 100644 index 0000000..35f917d --- /dev/null +++ b/tarot-lab/tests/test_db.py @@ -0,0 +1,70 @@ +import os +import pytest + +from app import db as db_module + + +@pytest.fixture(autouse=True) +def fresh_db(monkeypatch, tmp_path): + db_file = tmp_path / "test_tarot.db" + monkeypatch.setattr(db_module, "DB_PATH", str(db_file)) + db_module.init_db() + yield + try: + if db_file.exists(): + db_file.unlink() + except PermissionError: + pass # Windows SQLite WAL 잠금 + + +def test_save_and_get(): + rid = db_module.save_tarot_reading({ + "spread_type": "three_card", + "category": "연애", + "question": "Q", + "cards": [{"position": "과거", "card_id": "the-fool", "reversed": False}], + "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "A", "warning": None, "confidence": "medium"}, + "model": "claude-sonnet-4-6", + "tokens_in": 100, "tokens_out": 200, "cost_usd": 0.005, + "confidence": "medium", + }) + assert rid > 0 + row = db_module.get_tarot_reading(rid) + assert row["id"] == rid + assert row["category"] == "연애" + assert row["interpretation_json"]["summary"] == "S" + assert row["favorite"] == 0 + + +def test_list_with_filters(): + for cat in ["연애", "연애", "재물"]: + db_module.save_tarot_reading({ + "spread_type": "three_card", "category": cat, "question": "Q", + "cards": [], "interpretation_json": {"summary": "S", "cards": [], "interactions": [], "advice": "", "warning": None, "confidence": "low"}, + "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": "low", + }) + res = db_module.list_tarot_readings(page=1, size=10, category="연애") + assert res["total"] == 2 + assert all(r["category"] == "연애" for r in res["items"]) + + +def test_update_favorite_and_note(): + rid = db_module.save_tarot_reading({ + "spread_type": "one_card", "category": None, "question": None, + "cards": [], "interpretation_json": None, + "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": None, + }) + db_module.update_tarot_reading(rid, favorite=True, note="좋아요") + row = db_module.get_tarot_reading(rid) + assert row["favorite"] == 1 + assert row["note"] == "좋아요" + + +def test_delete(): + rid = db_module.save_tarot_reading({ + "spread_type": "one_card", "category": None, "question": None, + "cards": [], "interpretation_json": None, + "model": "x", "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0, "confidence": None, + }) + db_module.delete_tarot_reading(rid) + assert db_module.get_tarot_reading(rid) is None