Spaces:
Sleeping
Sleeping
File size: 1,879 Bytes
570f7bd 65f2e2c 570f7bd 4c2cf14 570f7bd 65f2e2c c1bc4eb 570f7bd 4c2cf14 65f2e2c 570f7bd 72e96d1 570f7bd 4c2cf14 570f7bd 4c2cf14 72e96d1 570f7bd 65f2e2c 570f7bd 65f2e2c 570f7bd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 |
import sqlite3
import logging
from typing import List, Tuple, Any
from adapters.db.base import DBAdapter
from pathlib import Path
log = logging.getLogger(__name__)
class SQLiteAdapter(DBAdapter):
name = "sqlite"
dialect = "sqlite"
def __init__(self, path: str):
# resolve absolute path for safety
self.path = Path(path).resolve()
log.info("SQLiteAdapter initialized with DB path: %s", self.path)
def preview_schema(self, limit_per_table: int = 0) -> str:
if not self.path.exists():
raise FileNotFoundError(f"SQLite DB does not exist: {self.path}")
with sqlite3.connect(f"file:{self.path}?mode=ro", uri=True) as conn:
cur = conn.cursor()
cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [t[0] for t in cur.fetchall()]
lines = []
for t in tables:
cur.execute(f"PRAGMA table_info({t});")
cols = [f"{c[1]}:{c[2]}" for c in cur.fetchall()]
lines.append(f"- {t} ({', '.join(cols)})")
return "\n".join(lines)
def execute(self, sql: str) -> Tuple[List[Tuple[Any, ...]], List[str]]:
if not self.path.exists():
raise FileNotFoundError(f"SQLite DB does not exist: {self.path}")
# use proper SQLite URI (not .as_uri())
uri = f"file:{self.path}?mode=ro"
log.info("SQLiteAdapter opening read-only connection to: %s", uri)
with sqlite3.connect(uri, uri=True, timeout=3) as conn:
cur = conn.cursor()
log.debug("Executing SQL: %s", sql.strip().replace("\n", " "))
cur.execute(sql)
rows = cur.fetchall()
cols = [desc[0] for desc in cur.description]
log.info("Query executed successfully. Returned %d rows.", len(rows))
return rows, cols
|