forked from Rockachopa/Timmy-time-dashboard
50
src/timmy/thinking/_db.py
Normal file
50
src/timmy/thinking/_db.py
Normal file
@@ -0,0 +1,50 @@
|
||||
"""Database models and access layer for the thinking engine."""
|
||||
|
||||
import sqlite3
|
||||
from collections.abc import Generator
|
||||
from contextlib import closing, contextmanager
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
_DEFAULT_DB = Path("data/thoughts.db")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Thought:
|
||||
"""A single thought in Timmy's inner stream."""
|
||||
|
||||
id: str
|
||||
content: str
|
||||
seed_type: str
|
||||
parent_id: str | None
|
||||
created_at: str
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _get_conn(db_path: Path = _DEFAULT_DB) -> Generator[sqlite3.Connection, None, None]:
|
||||
"""Get a SQLite connection with the thoughts table created."""
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with closing(sqlite3.connect(str(db_path))) as conn:
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("""
|
||||
CREATE TABLE IF NOT EXISTS thoughts (
|
||||
id TEXT PRIMARY KEY,
|
||||
content TEXT NOT NULL,
|
||||
seed_type TEXT NOT NULL,
|
||||
parent_id TEXT,
|
||||
created_at TEXT NOT NULL
|
||||
)
|
||||
""")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_thoughts_time ON thoughts(created_at)")
|
||||
conn.commit()
|
||||
yield conn
|
||||
|
||||
|
||||
def _row_to_thought(row: sqlite3.Row) -> Thought:
|
||||
return Thought(
|
||||
id=row["id"],
|
||||
content=row["content"],
|
||||
seed_type=row["seed_type"],
|
||||
parent_id=row["parent_id"],
|
||||
created_at=row["created_at"],
|
||||
)
|
||||
Reference in New Issue
Block a user