File size: 1,977 Bytes
98aa813 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """Exécuteur SQL lecture seule sur SQLite (company.db)."""
from __future__ import annotations
import logging
import sqlite3
import time
from pathlib import Path
from app.core.config import get_settings
from app.sql.seed import ensure_company_db
from app.sql.sql_validator import MAX_ROWS, is_safe_sql
logger = logging.getLogger(__name__)
def get_db_path() -> Path:
settings = get_settings()
return Path(settings.sqlite_db_path)
def _connect() -> sqlite3.Connection:
db_path = ensure_company_db(get_db_path())
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=5.0)
conn.row_factory = sqlite3.Row
return conn
def execute_readonly(query: str, params: tuple | list | None = None) -> dict:
"""Exécute un SELECT validé, avec limite de lignes et journal d'audit."""
params = tuple(params or ())
started = time.perf_counter()
if not is_safe_sql(query):
return {
"error": "Requête SQL refusée par le validateur.",
"query": query,
"duration_ms": 0,
}
limited = query.strip().rstrip(";")
if "LIMIT" not in limited.upper():
limited = f"{limited} LIMIT {MAX_ROWS}"
try:
with _connect() as conn:
cur = conn.execute(limited, params)
rows = [dict(r) for r in cur.fetchall()]
except sqlite3.Error as exc:
logger.warning("Échec SQL: %s", exc)
return {
"error": str(exc),
"query": limited,
"params": list(params),
"duration_ms": int((time.perf_counter() - started) * 1000),
}
duration_ms = int((time.perf_counter() - started) * 1000)
logger.info(
"SQL audit query=%r params=%s rows=%d duration_ms=%d",
limited, params, len(rows), duration_ms,
)
return {
"query": limited,
"params": list(params),
"rows": rows,
"row_count": len(rows),
"duration_ms": duration_ms,
}
|