| """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, |
| } |
|
|
