jpmschweitzer/settled-reach
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(db): extract shared DB module from duplicated patterns

Browse Source 
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>
This commit is contained in:
jpmschweitzer
2026-04-05 09:39:46 +02:00
co-authored by Claude Opus 4.6
parent 4dd866c9ad
commit d86ebada8d
6 changed files with 130 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+26
View File
@@ -0,0 +1,26 @@
[project]
name = "settled-reach-tooling"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
"PyYAML",
"jsonschema",
"numpy",
]
[project.optional-dependencies]
dev = [
# ruff 0.15.9 — checked clean against NVD + OSV, no CVEs on record (2026-04-05)
"ruff==0.15.9",
]
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
# E9xx: Runtime syntax/encoding errors
# F401: Unused imports
# F811: Redefinition of unused name
# F821: Undefined name (catches missing imports like bare `os`)
select = ["E9", "F401", "F811", "F821"]
tooling/db/common.py
+85
View File
@@ -0,0 +1,85 @@
#!/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)
tooling/db/decisions_sync.py
+4 -26
View File
@@ -17,38 +17,16 @@ import sqlite3
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import CONFIG_PATH, WORKTREE_ROOT, get_connection, load_config # noqa: E402
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql"
DECISIONS_DIR = WORKTREE_ROOT / "decisions"
# Database path: SR_DB_PATH env var (absolute), or fallback to parent directory heuristic.
DB_PATH = Path(os.environ["SR_DB_PATH"]).resolve() if os.environ.get("SR_DB_PATH") else (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
# ---------------------------------------------------------------------------
# Config / DB (same pattern as sqlite_connector.py)
# ---------------------------------------------------------------------------
def load_config():
"""Load config.json and resolve the SQLite database path."""
with open(CONFIG_PATH, "r") as f:
cfg = json.load(f)
cfg["sqlite_db_resolved"] = str(DB_PATH)
return cfg
def get_connection(cfg):
"""Return an sqlite3 connection with WAL mode and 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
# ---------------------------------------------------------------------------
tooling/db/sprint
+9 -19
View File
@@ -15,17 +15,16 @@ Usage:
"""
import json
import os
import sqlite3
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import WORKTREE_ROOT, get_connection, load_config # noqa: E402
SCRIPT_DIR = Path(__file__).resolve().parent
TICKET_CLI = str(SCRIPT_DIR / "ticket")
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
# Database path: SR_DB_PATH env var (absolute), or fallback to parent directory heuristic.
DB_PATH = Path(os.environ["SR_DB_PATH"]).resolve() if os.environ.get("SR_DB_PATH") else (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
PROJECT_ROOT = WORKTREE_ROOT
REMINDER = """---
@@ -48,15 +47,6 @@ def run_ticket(*args):
return {"ok": False, "error": f"Bad ticket output: {result.stdout[:200]}"}
def get_connection():
"""Direct DB connection for lifecycle mutations only."""
conn = sqlite3.connect(str(DB_PATH))
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA foreign_keys=ON;")
conn.row_factory = sqlite3.Row
return conn
def parse_flags(args, known_flags):
"""Parse --flag value pairs from args, return (flags_dict, positional_args)."""
flags = {}
@@ -325,7 +315,7 @@ def cmd_start(args):
sys.exit(1)
# Activate
conn = get_connection()
conn = get_connection(load_config())
conn.execute(
"UPDATE sprints SET status='active', start_date=date('now') WHERE id=?",
(sprint["id"],)
@@ -369,7 +359,7 @@ def cmd_stop(args):
picked_up = [t for t in incomplete if t["status"] in picked_up_statuses]
auto_closed = []
if picked_up:
conn = get_connection()
conn = get_connection(load_config())
for t in picked_up:
conn.execute("UPDATE tickets SET status='done' WHERE id=?", (t["id"],))
auto_closed.append(t)
@@ -380,7 +370,7 @@ def cmd_stop(args):
incomplete = [t for t in incomplete if t["status"] not in picked_up_statuses]
# Complete the sprint
conn = get_connection()
conn = get_connection(load_config())
conn.execute(
"UPDATE sprints SET status='completed', end_date=date('now') WHERE id=?",
(sprint["id"],)
@@ -602,7 +592,7 @@ def cmd_prepare(args):
# Create sprint record if it doesn't exist
if sprint.get("status") == "new":
conn = get_connection()
conn = get_connection(load_config())
conn.execute(
"INSERT INTO sprints (id, name, status) VALUES (?, ?, 'planning')",
(sprint["id"], f"Sprint {sprint['id']}")
@@ -664,7 +654,7 @@ def cmd_prepare(args):
print()
# Decision coverage gaps
conn = get_connection()
conn = get_connection(load_config())
cursor = conn.execute("""
SELECT id, title FROM decisions
WHERE type='confirmed' AND status='active'
tooling/db/sqlite_connector.py
+4 -28
View File
@@ -11,38 +11,15 @@ Usage:
"""
import json
import os
import sqlite3
import sys
from pathlib import Path
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import WORKTREE_ROOT, get_connection, load_config # noqa: E402
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql"
# Database path: SR_DB_PATH env var (absolute), or fallback to parent directory heuristic.
DB_PATH = Path(os.environ["SR_DB_PATH"]).resolve() if os.environ.get("SR_DB_PATH") else (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
def load_config():
"""Load config.json and resolve the SQLite database path."""
with open(CONFIG_PATH, "r") as f:
cfg = json.load(f)
cfg["sqlite_db_resolved"] = str(DB_PATH)
return cfg
def get_connection(cfg):
"""Return an sqlite3 connection with WAL mode and 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
# ---------------------------------------------------------------------------
@@ -180,9 +157,8 @@ Usage:
All output is JSON on stdout. Errors also use JSON with {{"ok": false, "error": "..."}}.
Config: {config}
Schema: {schema}
""".format(config=CONFIG_PATH, schema=SCHEMA_PATH)
""".format(schema=SCHEMA_PATH)
def main():
tooling/db/ticket
+2 -21
View File
@@ -23,31 +23,12 @@ All output is JSON on stdout.
"""
import json
import os
import sqlite3
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
# Database path: SR_DB_PATH env var (absolute), or fallback to parent directory heuristic.
DB_PATH = Path(os.environ["SR_DB_PATH"]).resolve() if os.environ.get("SR_DB_PATH") else (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
sys.path.insert(0, str(Path(__file__).resolve().parent))
def load_config():
with open(CONFIG_PATH, "r") as f:
cfg = json.load(f)
cfg["sqlite_db_resolved"] = str(DB_PATH)
return cfg
def get_connection(cfg):
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
from common import get_connection, load_config # noqa: E402
def query(conn, sql, params=()):