115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
import sys
|
|
from pathlib import Path
|
|
try:
|
|
from sqlcipher3 import dbapi2 as sqlite
|
|
|
|
SQLCIPHER_AVAILABLE = True
|
|
except ImportError:
|
|
import sqlite3 as sqlite
|
|
|
|
SQLCIPHER_AVAILABLE = False
|
|
|
|
sys.path.append(str(Path(__file__).resolve().parent.parent.parent))
|
|
from .config import DATA_DIR, DATABASE_FILENAME
|
|
from .models import JournalEntry
|
|
from .encryption import derive_key
|
|
|
|
|
|
def get_db_connection(password: str) -> sqlite.Connection:
|
|
"""
|
|
Creates and returns a connection to the encrypted SQLite database.
|
|
The database key is derived from the user's main vault password.
|
|
"""
|
|
db_path = DATA_DIR / DATABASE_FILENAME
|
|
# Use a fixed salt for the DB key so it's the same for the session.
|
|
# This is secure because the salt is only used with the user's high-entropy password.
|
|
db_salt = b"a_fixed_salt_for_the_db_key_deriv"
|
|
db_key = derive_key(password, db_salt)
|
|
|
|
conn = sqlite.connect(str(db_path))
|
|
if SQLCIPHER_AVAILABLE:
|
|
# The key must be provided as a hex string.
|
|
_ = conn.execute(f"PRAGMA key = \"x'{db_key.hex()}'\"")
|
|
# Test the connection to ensure the key is correct.
|
|
_ = conn.execute("SELECT count(*) FROM sqlite_master;")
|
|
else:
|
|
print(
|
|
"WARNING: sqlcipher3 is unavailable; using sqlite3 fallback without DB encryption."
|
|
)
|
|
return conn
|
|
|
|
|
|
def create_schema(conn: sqlite.Connection):
|
|
"""Creates the database schema if it doesn't already exist."""
|
|
cursor = conn.cursor()
|
|
# Entries Table
|
|
_ = cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS entries (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
date TEXT NOT NULL UNIQUE
|
|
)
|
|
"""
|
|
)
|
|
# Sections Table
|
|
_ = cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS sections (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
entry_id INTEGER NOT NULL,
|
|
title TEXT NOT NULL,
|
|
content TEXT,
|
|
FOREIGN KEY (entry_id) REFERENCES entries (id)
|
|
)
|
|
"""
|
|
)
|
|
# Fragments Table
|
|
_ = cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS fragments (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
entry_id INTEGER NOT NULL,
|
|
type TEXT NOT NULL,
|
|
description TEXT,
|
|
time TEXT,
|
|
FOREIGN KEY (entry_id) REFERENCES entries (id)
|
|
)
|
|
"""
|
|
)
|
|
# Tags Table
|
|
_ = cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS tags (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE
|
|
)
|
|
"""
|
|
)
|
|
# Fragment-Tags Join Table
|
|
_ = cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS fragment_tags (
|
|
fragment_id INTEGER NOT NULL,
|
|
tag_id INTEGER NOT NULL,
|
|
PRIMARY KEY (fragment_id, tag_id),
|
|
FOREIGN KEY (fragment_id) REFERENCES fragments (id),
|
|
FOREIGN KEY (tag_id) REFERENCES tags (id)
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def hydrate_database(conn: sqlite.Connection, entries: list[JournalEntry]):
|
|
"""
|
|
Populates the database with a list of JournalEntry objects.
|
|
This function is designed to be idempotent but is typically run on a clean DB.
|
|
"""
|
|
# This is a placeholder for the full hydration logic.
|
|
# A complete implementation would iterate through entries, sections, and fragments,
|
|
# inserting them into their respective tables.
|
|
print(f"Hydrating database with {len(entries)} entries...")
|
|
# For now, we just ensure the schema is created.
|
|
create_schema(conn)
|
|
print("Database hydration complete.")
|