nngoandClaude Sonnet 4.6
04476bae11
execute_script() semicolon splitter (all drivers)
- Add BaseDriver._split_statements() that tracks single/double-quoted
strings and -- / /* */ comments so semicolons inside procedure bodies
are not treated as statement boundaries.
- Replace the naive sql.split(';') in all four drivers with this helper.
- MSSQL execute_script() also pre-splits on GO (case-insensitive, own
line) so scripts pasted from SSMS work correctly.
MSSQL EXPLAIN
- SET SHOWPLAN_TEXT ON/execute/SET SHOWPLAN_TEXT OFF must be separate
execute() calls; pyodbc rejects multiple statements in one call.
Hold self._lock for the entire sequence to keep session state atomic.
Keyring (connections.py)
- Log warnings (with traceback) on load/save failures instead of
silently returning "".
- save_password() now calls delete_password() when password is empty,
so clearing a saved password actually removes the old keyring entry
rather than leaving a stale one behind.
SQLite get_tables() O(N) lock acquisitions
- Run all COUNT(*) queries inside the same `with self._cur() as c:`
block, reducing N+1 lock acquisitions to 1.
Unused import: remove psycopg2.extras from postgres_driver.py.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
107 lines
3.1 KiB
Python
107 lines
3.1 KiB
Python
"""
|
|
Connection profile persistence.
|
|
Profiles are stored as JSON in ~/.dbclient/connections.json.
|
|
Passwords are stored separately in the OS keychain via keyring.
|
|
"""
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from app.models.connection_model import ConnectionProfile
|
|
from app.utils.logger import get_logger
|
|
|
|
_log = get_logger(__name__)
|
|
|
|
_CONN_FILE = Path.home() / ".dbclient" / "connections.json"
|
|
|
|
try:
|
|
import keyring
|
|
_KEYRING_OK = True
|
|
except ImportError:
|
|
_KEYRING_OK = False
|
|
|
|
_SERVICE = "DBClient"
|
|
|
|
|
|
# ── Keyring helpers ──────────────────────────────────────────────────────────
|
|
|
|
def save_password(profile_id: str, password: str) -> None:
|
|
if not _KEYRING_OK:
|
|
return
|
|
if password:
|
|
try:
|
|
keyring.set_password(_SERVICE, profile_id, password)
|
|
except Exception:
|
|
_log.warning("keyring: failed to save password for %s", profile_id, exc_info=True)
|
|
else:
|
|
delete_password(profile_id)
|
|
|
|
|
|
def load_password(profile_id: str) -> str:
|
|
if _KEYRING_OK:
|
|
try:
|
|
return keyring.get_password(_SERVICE, profile_id) or ""
|
|
except Exception:
|
|
_log.warning("keyring: failed to load password for %s", profile_id, exc_info=True)
|
|
return ""
|
|
return ""
|
|
|
|
|
|
def delete_password(profile_id: str) -> None:
|
|
if _KEYRING_OK:
|
|
try:
|
|
keyring.delete_password(_SERVICE, profile_id)
|
|
except Exception:
|
|
pass # Already absent is fine
|
|
|
|
|
|
# ── Profile CRUD ─────────────────────────────────────────────────────────────
|
|
|
|
def _load_raw() -> list:
|
|
if _CONN_FILE.exists():
|
|
try:
|
|
with open(_CONN_FILE, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return []
|
|
return []
|
|
|
|
|
|
def _save_raw(data: list) -> None:
|
|
_CONN_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(_CONN_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
def load_profiles() -> list:
|
|
"""Return list of ConnectionProfile with passwords injected from keyring."""
|
|
profiles = []
|
|
for raw in _load_raw():
|
|
p = ConnectionProfile.from_dict(raw)
|
|
p.password = load_password(p.id)
|
|
profiles.append(p)
|
|
return profiles
|
|
|
|
|
|
def save_profile(profile: ConnectionProfile) -> None:
|
|
"""Upsert a profile (save or update)."""
|
|
raw_list = _load_raw()
|
|
# Replace existing or append
|
|
found = False
|
|
for i, raw in enumerate(raw_list):
|
|
if raw.get("id") == profile.id:
|
|
raw_list[i] = profile.to_dict()
|
|
found = True
|
|
break
|
|
if not found:
|
|
raw_list.append(profile.to_dict())
|
|
_save_raw(raw_list)
|
|
save_password(profile.id, profile.password)
|
|
|
|
|
|
def delete_profile(profile_id: str) -> None:
|
|
"""Remove a profile by ID."""
|
|
raw_list = [r for r in _load_raw() if r.get("id") != profile_id]
|
|
_save_raw(raw_list)
|
|
delete_password(profile_id)
|