"""Database connection management for iMessage chat.db."""
import os
import sqlite3
from contextlib import contextmanager
from typing import Generator
from ..constants import CHAT_DB_PATH, ATTACHMENTS_PATH
from ..exceptions import PermissionDeniedError, DatabaseLockedError
def check_database_access() -> dict[str, bool | str]:
"""Check if we have access to the iMessage database and attachments.
Returns:
Dictionary with access status for each path and any error messages.
"""
result: dict[str, bool | str] = {
"chat_db_exists": os.path.exists(CHAT_DB_PATH),
"chat_db_readable": os.access(CHAT_DB_PATH, os.R_OK),
"attachments_exists": os.path.exists(ATTACHMENTS_PATH),
"attachments_readable": os.access(ATTACHMENTS_PATH, os.R_OK),
"chat_db_path": CHAT_DB_PATH,
"attachments_path": ATTACHMENTS_PATH,
}
if not result["chat_db_readable"]:
result["error"] = (
"Cannot read chat.db. Please grant Full Disk Access to "
"Terminal.app or Claude Desktop.app in System Settings > "
"Privacy & Security > Full Disk Access."
)
return result
@contextmanager
def get_connection(
timeout: float = 5.0,
) -> Generator[sqlite3.Connection, None, None]:
"""Get a read-only connection to the iMessage database.
Uses WAL-aware connection parameters for safe concurrent access
while Messages.app may be writing to the database.
Args:
timeout: Busy timeout in seconds (how long to wait if DB is locked).
Yields:
SQLite connection configured for read-only access.
Raises:
PermissionDeniedError: If Full Disk Access is not granted.
DatabaseLockedError: If the database remains locked after timeout.
"""
if not os.access(CHAT_DB_PATH, os.R_OK):
raise PermissionDeniedError(
f"Cannot read {CHAT_DB_PATH}. Please grant Full Disk Access to "
"Terminal.app or Claude Desktop.app."
)
# Use URI mode for read-only access
# mode=ro ensures we don't accidentally modify the database
uri = f"file:{CHAT_DB_PATH}?mode=ro"
try:
conn = sqlite3.connect(
uri,
uri=True,
timeout=timeout,
check_same_thread=False,
)
# Enable row factory for dict-like access
conn.row_factory = sqlite3.Row
# Set busy timeout in milliseconds
conn.execute(f"PRAGMA busy_timeout = {int(timeout * 1000)}")
yield conn
except sqlite3.OperationalError as e:
error_msg = str(e).lower()
if "locked" in error_msg or "busy" in error_msg:
raise DatabaseLockedError(
"chat.db is locked by Messages.app. Please try again."
) from e
elif "unable to open" in error_msg or "permission" in error_msg:
raise PermissionDeniedError(
f"Cannot open {CHAT_DB_PATH}. Please grant Full Disk Access."
) from e
else:
raise
finally:
if "conn" in locals():
conn.close()
