adapters/db/sqlite_adapter.py

import sqlite3
from typing import List, Tuple, Any
from adapters.db.base import DBAdapter


class SQLiteAdapter(DBAdapter):
    name = "sqlite"
    dialect = "sqlite"

    def __init__(self, path: str):
        self.path = path

    def preview_schema(self, limit_per_table: int = 0) -> str:
        with sqlite3.connect(self.path, uri=True) as conn:
            cur = conn.cursor()
            cur.execute("PRAGMA foreign_keys = ON")
            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]]:
        # enforce read-only connection
        uri = f"file:{self.path}?mode=ro&uri=true"
        with sqlite3.connect(uri, uri=True, timeout=3) as conn:
            cur = conn.cursor()
            cur.execute(sql)
            rows = cur.fetchall()
            cols = [desc[0] for desc in cur.description]
            return rows, cols