Ottimizza barcode e disattiva profiler query

async_msssql_query.py

@@ -10,9 +10,13 @@ cross-loop errors.
 from __future__ import annotations
 
 import asyncio
+import inspect
+import json
 import logging
+import os
 import time
 import urllib.parse
+from pathlib import Path
 from typing import Any, Dict, Optional
 
 from sqlalchemy import text
@@ -22,6 +26,59 @@ from version_info import module_version
 
 __version__ = module_version(__name__)
 
+QUERY_PROFILER_LOG = Path(__file__).with_name("warehouse_query_profiler.log")
+QUERY_PROFILER_CONFIG_PATH = Path(__file__).with_name("db_connection.json")
+
+
+def _load_query_profiler_config() -> dict[str, Any]:
+    """Read profiler settings from db_connection.json, with env overrides."""
+
+    config: dict[str, Any] = {
+        "enabled": False,
+        "slow_ms": 0,
+        "sql_limit": 4000,
+    }
+    try:
+        data = json.loads(QUERY_PROFILER_CONFIG_PATH.read_text(encoding="utf-8"))
+        section = data.get("query_profiler") if isinstance(data, dict) else None
+        if isinstance(section, dict):
+            config.update(section)
+    except Exception:
+        pass
+    if "WAREHOUSE_QUERY_PROFILER" in os.environ:
+        config["enabled"] = os.environ.get("WAREHOUSE_QUERY_PROFILER", "1").strip().lower() not in {
+            "0",
+            "false",
+            "no",
+            "off",
+        }
+    if "WAREHOUSE_QUERY_PROFILER_SLOW_MS" in os.environ:
+        config["slow_ms"] = os.environ.get("WAREHOUSE_QUERY_PROFILER_SLOW_MS", "0")
+    if "WAREHOUSE_QUERY_PROFILER_SQL_LIMIT" in os.environ:
+        config["sql_limit"] = os.environ.get("WAREHOUSE_QUERY_PROFILER_SQL_LIMIT", "4000")
+    return config
+
+
+def _query_profiler_enabled() -> bool:
+    value = _load_query_profiler_config().get("enabled", False)
+    if isinstance(value, str):
+        return value.strip().lower() not in {"0", "false", "no", "off"}
+    return bool(value)
+
+
+def _query_profiler_slow_ms() -> float:
+    try:
+        return float(_load_query_profiler_config().get("slow_ms", 0) or 0)
+    except Exception:
+        return 0.0
+
+
+def _query_profiler_sql_limit() -> int:
+    try:
+        return int(_load_query_profiler_config().get("sql_limit", 4000) or 4000)
+    except Exception:
+        return 4000
+
 try:
     import pyodbc
 
@@ -76,6 +133,83 @@ def make_mssql_dsn(
     return f"mssql+aioodbc:///?odbc_connect={urllib.parse.quote_plus(odbc)}"
 
 
+def _compact_sql(sql: str, *, limit: int | None = None) -> str:
+    """Collapse SQL whitespace so profiler entries stay readable."""
+
+    if limit is None:
+        limit = _query_profiler_sql_limit()
+    text_value = " ".join(str(sql or "").split())
+    if limit > 0 and len(text_value) > limit:
+        return text_value[:limit] + "...<truncated>"
+    return text_value
+
+
+def _profile_params(params: Optional[Dict[str, Any]]) -> str:
+    """Serialize SQL parameters for diagnostics."""
+
+    if not params:
+        return "{}"
+    try:
+        return _dumps(params)
+    except Exception:
+        return repr(params)
+
+
+def _query_caller() -> str:
+    """Return the first external Python frame that triggered the DB call."""
+
+    current_file = Path(__file__).resolve()
+    for frame in inspect.stack(context=0)[2:]:
+        try:
+            frame_file = Path(frame.filename).resolve()
+        except Exception:
+            continue
+        if frame_file == current_file:
+            continue
+        return f"{frame_file.name}:{frame.lineno}:{frame.function}"
+    return "unknown"
+
+
+def _append_query_profile(
+    *,
+    method: str,
+    elapsed_ms: float,
+    rows: int | None,
+    rowcount: int | None,
+    commit: bool,
+    ok: bool,
+    sql: str,
+    params: Optional[Dict[str, Any]],
+    caller: str,
+    error: str = "",
+) -> None:
+    """Append one query timing line to the local profiler log."""
+
+    if not _query_profiler_enabled():
+        return
+    slow_ms = _query_profiler_slow_ms()
+    if ok and slow_ms > 0 and elapsed_ms < slow_ms:
+        return
+    try:
+        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
+        row_info = f"rows={rows}" if rows is not None else f"rowcount={rowcount}"
+        lines = [
+            (
+                f"{timestamp} | {elapsed_ms:.3f} ms | {method} | "
+                f"{row_info} | commit={int(commit)} | ok={int(ok)} | caller={caller}"
+            ),
+            f"PARAMS: {_profile_params(params)}",
+            f"SQL: {_compact_sql(sql)}",
+        ]
+        if error:
+            lines.append(f"ERROR: {error}")
+        with QUERY_PROFILER_LOG.open("a", encoding="utf-8") as handle:
+            handle.write("\n".join(lines) + "\n---\n")
+    except Exception:
+        # Profiling must never interfere with warehouse operations.
+        pass
+
+
 class AsyncMSSQLClient:
     """Thin async query client for SQL Server.
 
@@ -152,23 +286,83 @@ class AsyncMSSQLClient:
         """
         await self._ensure_engine()
         t0 = time.perf_counter()
-        async with (self._engine.begin() if commit else self._engine.connect()) as conn:
-            res = await conn.execute(text(sql), params or {})
-            rows = res.fetchall()
-            cols = list(res.keys())
+        caller = _query_caller()
+        try:
+            async with (self._engine.begin() if commit else self._engine.connect()) as conn:
+                res = await conn.execute(text(sql), params or {})
+                rows = res.fetchall()
+                cols = list(res.keys())
+        except Exception as exc:
+            elapsed_ms = round((time.perf_counter() - t0) * 1000, 3)
+            _append_query_profile(
+                method="query_json",
+                elapsed_ms=elapsed_ms,
+                rows=None,
+                rowcount=None,
+                commit=commit,
+                ok=False,
+                sql=sql,
+                params=params,
+                caller=caller,
+                error=repr(exc),
+            )
+            raise
         if as_dict_rows:
             rows_out = [dict(zip(cols, row)) for row in rows]
         else:
             rows_out = [list(row) for row in rows]
+        elapsed_ms = round((time.perf_counter() - t0) * 1000, 3)
+        _append_query_profile(
+            method="query_json",
+            elapsed_ms=elapsed_ms,
+            rows=len(rows_out),
+            rowcount=None,
+            commit=commit,
+            ok=True,
+            sql=sql,
+            params=params,
+            caller=caller,
+        )
         return {
             "columns": cols,
             "rows": rows_out,
-            "elapsed_ms": round((time.perf_counter() - t0) * 1000, 3),
+            "elapsed_ms": elapsed_ms,
         }
 
     async def exec(self, sql: str, params: Optional[Dict[str, Any]] = None, *, commit: bool = False) -> int:
         """Execute a DML statement and return its row count."""
         await self._ensure_engine()
-        async with (self._engine.begin() if commit else self._engine.connect()) as conn:
-            res = await conn.execute(text(sql), params or {})
-            return res.rowcount or 0
+        t0 = time.perf_counter()
+        caller = _query_caller()
+        try:
+            async with (self._engine.begin() if commit else self._engine.connect()) as conn:
+                res = await conn.execute(text(sql), params or {})
+                rowcount = res.rowcount or 0
+        except Exception as exc:
+            elapsed_ms = round((time.perf_counter() - t0) * 1000, 3)
+            _append_query_profile(
+                method="exec",
+                elapsed_ms=elapsed_ms,
+                rows=None,
+                rowcount=None,
+                commit=commit,
+                ok=False,
+                sql=sql,
+                params=params,
+                caller=caller,
+                error=repr(exc),
+            )
+            raise
+        elapsed_ms = round((time.perf_counter() - t0) * 1000, 3)
+        _append_query_profile(
+            method="exec",
+            elapsed_ms=elapsed_ms,
+            rows=None,
+            rowcount=rowcount,
+            commit=commit,
+            ok=True,
+            sql=sql,
+            params=params,
+            caller=caller,
+        )
+        return rowcount