File size: 3,284 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """Construction de requêtes SQL paramétrées (pas de SQL libre LLM)."""
from __future__ import annotations
from typing import Any, Literal
from app.sql.executor import execute_readonly
from app.sql.sql_validator import assert_whitelisted_identifiers
AGG_SQL = {
"sum": "SUM",
"max": "MAX",
"min": "MIN",
"mean": "AVG",
"avg": "AVG",
"count": "COUNT",
}
def build_sales_query(
metric: Literal["chiffre_affaires", "marge"] = "chiffre_affaires",
aggregation: str = "sum",
filters: dict[str, Any] | None = None,
group_by: str | None = None,
top_n: int | None = None,
) -> tuple[str, list[Any]]:
"""Construit un SELECT whitelisté sur la table sales."""
filters = filters or {}
agg = AGG_SQL.get(aggregation.lower())
if agg is None:
raise ValueError(f"Agrégation inconnue: {aggregation}")
columns = [metric, *filters.keys()]
if group_by:
columns.append(group_by)
assert_whitelisted_identifiers("sales", columns, agg)
where_parts: list[str] = []
params: list[Any] = []
for key, value in filters.items():
where_parts.append(f"LOWER({key}) = LOWER(?)")
params.append(value)
where_sql = f" WHERE {' AND '.join(where_parts)}" if where_parts else ""
if group_by:
query = (
f"SELECT {group_by} AS label, {agg}({metric}) AS value "
f"FROM sales{where_sql} "
f"GROUP BY {group_by} "
f"ORDER BY value DESC"
)
if top_n:
query += f" LIMIT {int(top_n)}"
else:
query = (
f"SELECT {agg}({metric}) AS value, COUNT(*) AS rows_considered "
f"FROM sales{where_sql}"
)
return query, params
def run_sql_query(
metric: Literal["chiffre_affaires", "marge"] = "chiffre_affaires",
aggregation: str = "sum",
filters: dict | None = None,
group_by: str | None = None,
top_n: int | None = None,
) -> dict:
"""Point d'entrée métier : construit + exécute + normalise le résultat."""
try:
query, params = build_sales_query(
metric=metric,
aggregation=aggregation,
filters=filters,
group_by=group_by,
top_n=top_n,
)
except ValueError as exc:
return {"error": str(exc)}
raw = execute_readonly(query, params)
if "error" in raw:
return raw
rows = raw.get("rows", [])
if group_by:
ranked = [
(r["label"], float(r["value"]) if r["value"] is not None else 0.0)
for r in rows
]
return {
"metric": metric,
"aggregation": aggregation,
"group_by": group_by,
"filters": filters or {},
"result": ranked,
"sql": raw.get("query"),
"duration_ms": raw.get("duration_ms"),
}
value = rows[0]["value"] if rows else 0
considered = rows[0].get("rows_considered", 0) if rows else 0
return {
"metric": metric,
"aggregation": aggregation,
"filters": filters or {},
"result": float(value) if value is not None else 0.0,
"rows_considered": considered,
"sql": raw.get("query"),
"duration_ms": raw.get("duration_ms"),
}
|