jpmschweitzerandClaude Opus 4.6
d86ebada8d
Create tooling/db/common.py with resolve_db_path(), load_config(), get_connection(), and ensure_venv(). Update ticket, sprint, sqlite_connector.py, and decisions_sync.py to import from common instead of duplicating. Fixes pre-existing NameError in decisions_sync.py (missing import os). Add pyproject.toml with ruff config (ruff==0.15.9, CVE-clean) and dev dependencies. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
86 lines
2.7 KiB
Python
86 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared utilities for Settled Reach DB tooling scripts.
|
|
|
|
Provides the standard DB path resolution, config loading, and connection
|
|
helpers used across ticket, sprint, sqlite_connector, and decisions_sync.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
CONFIG_PATH = SCRIPT_DIR / "config.json"
|
|
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def resolve_db_path() -> Path:
|
|
"""Return the ticket database path.
|
|
|
|
Prefers the SR_DB_PATH environment variable (absolute path).
|
|
Falls back to <worktree-parent>/settledreach.db — the shared DB location
|
|
used when working directly in the main repo checkout.
|
|
"""
|
|
if os.environ.get("SR_DB_PATH"):
|
|
return Path(os.environ["SR_DB_PATH"]).resolve()
|
|
return (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
|
|
|
|
|
|
def load_config() -> dict:
|
|
"""Load tooling/db/config.json and inject the resolved DB path."""
|
|
with open(CONFIG_PATH) as f:
|
|
cfg = json.load(f)
|
|
cfg["sqlite_db_resolved"] = str(resolve_db_path())
|
|
return cfg
|
|
|
|
|
|
def get_connection(cfg: dict) -> sqlite3.Connection:
|
|
"""Return a WAL-mode SQLite connection with foreign keys enabled."""
|
|
conn = sqlite3.connect(cfg["sqlite_db_resolved"])
|
|
conn.execute("PRAGMA journal_mode=WAL;")
|
|
conn.execute("PRAGMA foreign_keys=ON;")
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def ensure_venv() -> None:
|
|
"""Re-exec into the project .venv Python if not already running there.
|
|
|
|
Call this at the top of any script that uses third-party packages,
|
|
before those imports. Pure-stdlib scripts (ticket, sprint,
|
|
sqlite_connector, decisions_sync) do not need it.
|
|
|
|
Usage::
|
|
|
|
from common import ensure_venv
|
|
ensure_venv()
|
|
import numpy as np # third-party import follows
|
|
"""
|
|
venv_python = WORKTREE_ROOT / ".venv" / "bin" / "python"
|
|
|
|
# Already running inside the venv — nothing to do.
|
|
if Path(sys.executable).resolve() == venv_python.resolve():
|
|
return
|
|
|
|
# .venv not set up yet — fail with a helpful message.
|
|
if not venv_python.exists():
|
|
print(
|
|
"error: .venv not found — run `make setup-venv` first.",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Re-exec into the venv Python, preserving all arguments.
|
|
os.execv(str(venv_python), [str(venv_python)] + sys.argv)
|