refactor(backup): route read-only restore opens through read_only_db_uri

The three percent-encoded mode=ro connects added earlier in this stack
re-implemented hermes_state_holders.read_only_db_uri, which already exists
for exactly this '#'/'?' truncation bug and is the canonical builder used by
hermes_state and doctor_state. Using it keeps a single definition of the
encoding so a future fix lands everywhere. Behaviour is unchanged
(timeout=1.0 kept at the backup.py site).
This commit is contained in:
kshitijk4poor
2026-09-26 19:35:59 +05:30
committed by kshitij
parent 1d9f03b7b6
commit 48f1bf7251
2 changed files with 5 additions and 3 deletions

View File

@@ -22,6 +22,7 @@ from hermes_constants import (
display_hermes_home,
)
from hermes_state_dbfile import RETIRED_GENERATION_DIR_SUFFIX
from hermes_state_holders import read_only_db_uri
from hermes_cli.archive_safe import normalize_archive_parts
from hermes_cli.backup_sqlite import _close_quietly, _safe_copy_db
@@ -348,7 +349,7 @@ def _query_ro_sqlite(path: Path, fn):
"""Run ``fn(conn)`` on a read-only connection to *path*; return ``(value, None)`` or ``(None, exc)``."""
conn = None
try:
conn = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True, timeout=1.0)
conn = sqlite3.connect(read_only_db_uri(path), uri=True, timeout=1.0)
return fn(conn), None
except Exception as exc:
return None, exc

View File

@@ -18,6 +18,7 @@ import zipfile
from pathlib import Path
from typing import List, Optional, Tuple
from hermes_state_holders import read_only_db_uri
from utils import (
_preserve_file_mode, _preserve_file_owner, _restore_file_mode, _restore_file_owner, atomic_replace,
)
@@ -116,7 +117,7 @@ def _safe_restore_db(src: Path, dst: Path) -> bool:
dst_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
except Exception:
pass
src_conn = sqlite3.connect(f"{src.resolve().as_uri()}?mode=ro", uri=True)
src_conn = sqlite3.connect(read_only_db_uri(src), uri=True)
try:
src_conn.backup(dst_conn)
finally:
@@ -390,7 +391,7 @@ def _count_session_rows(path: Path) -> Optional[Tuple[int, int]]:
if not path.is_file():
return None
try:
conn = sqlite3.connect(f"{path.resolve().as_uri()}?mode=ro", uri=True)
conn = sqlite3.connect(read_only_db_uri(path), uri=True)
except sqlite3.Error:
return None
try: