"""SQLite database initialization and connection management."""
import sqlite3
from contextlib import contextmanager
from pathlib import Path
from typing import Generator
from ..config import DATABASE_PATH
def init_database() -> None:
"""Initialize the SQLite database with required tables."""
DATABASE_PATH.parent.mkdir(parents=True, exist_ok=True)
with get_db_connection() as conn:
cursor = conn.cursor()
# Change log table
cursor.execute("""
CREATE TABLE IF NOT EXISTS change_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
entity_type TEXT NOT NULL,
entity_id TEXT,
action TEXT NOT NULL,
details TEXT,
user_context TEXT
)
""")
# Report history table
cursor.execute("""
CREATE TABLE IF NOT EXISTS report_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
report_type TEXT NOT NULL,
date_range TEXT NOT NULL,
filters TEXT,
data TEXT NOT NULL,
summary TEXT
)
""")
# Decisions table
cursor.execute("""
CREATE TABLE IF NOT EXISTS decisions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
decision_type TEXT NOT NULL,
description TEXT NOT NULL,
rationale TEXT NOT NULL,
alternatives_considered TEXT,
outcome TEXT,
related_entities TEXT
)
""")
# Learnings table
cursor.execute("""
CREATE TABLE IF NOT EXISTS learnings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
category TEXT NOT NULL,
insight TEXT NOT NULL,
evidence TEXT,
confidence TEXT DEFAULT 'medium',
actionable_recommendations TEXT,
related_decisions TEXT
)
""")
# Create indexes for common queries
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_change_log_timestamp
ON change_log(timestamp DESC)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_report_history_type
ON report_history(report_type, timestamp DESC)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_decisions_type
ON decisions(decision_type, timestamp DESC)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_learnings_category
ON learnings(category, timestamp DESC)
""")
conn.commit()
@contextmanager
def get_db_connection() -> Generator[sqlite3.Connection, None, None]:
"""Get a database connection with automatic cleanup.
Yields:
SQLite connection object.
"""
conn = sqlite3.connect(DATABASE_PATH)
conn.row_factory = sqlite3.Row # Enable dict-like access to rows
try:
yield conn
finally:
conn.close()
