perf(db): batch the good-parts N+1 + cache the press/reason reference reads

- _good_parts_map: collapse the per-press 3-query loop (3xN) into 3 grouped
  queries across all presses; semantics preserved (Joborder JSON/substring
  matching, per-press scoping, whole-map degrade-to-{} on error).
- _press_list + /reasons: served from the kernel cached_select() TTL cache.

Part of the gateway connection-pool-saturation fix
(see fastAPI/agent_docs/db-pool-relief.md).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

_db.py

@@ -18,9 +18,9 @@ from __future__ import annotations
 
 from contextlib import contextmanager
 
-from app.db.mariadb import engine, run_select_query  # noqa: F401  (kernel DB layer)
+from app.db.mariadb import engine, run_select_query, cached_select  # noqa: F401  (kernel DB layer)
 
-__all__ = ["run_select_query", "transaction", "engine"]
+__all__ = ["run_select_query", "cached_select", "transaction", "engine"]
 
 
 @contextmanager

api/routes.py

@@ -51,7 +51,7 @@ from pydantic import BaseModel, ConfigDict
 from sqlalchemy import text
 from starlette import status
 
-from .._db import run_select_query, transaction
+from .._db import run_select_query, cached_select, transaction
 
 logger = logging.getLogger(__name__)
 
@@ -96,7 +96,9 @@ def _press_list(plant: str, entry_presses: set[str]) -> list[str]:
     if the registry is unavailable. Numeric press ids sort numerically."""
     presses: set[str] = set(entry_presses)
     try:
-        rows = run_select_query(
+        # Reference registry (admin-edited); served from cache so it's off the pool each poll.
+        rows = cached_select(
+            f"trj:presses:{plant}",
             "SELECT psPress AS p FROM press_settings.presses WHERE psPlant = :plant",
             {"plant": plant},
         )
@@ -172,51 +174,58 @@ def _good_parts_map(plant: str, pairs: set[tuple[str, str]]) -> dict[tuple[str,
         return {}
     import json
 
-    by_press: dict[str, list[str]] = defaultdict(list)
+    # Batched across ALL presses: 3 grouped queries total (was 3 per press = N+1).
-    for p, o in pairs:
+    # Semantics preserved from the per-press version — only the query scope changed
-        by_press[p].append(o)
+    # (grouped by press). On any DB error the whole map degrades to {} (the caller's
-
+    # _flow_for then treats notYetSent as 0), matching the sibling _temper_agg_map.
-    result: dict[tuple[str, str], int] = {}
+    presses = sorted({p for p, _ in pairs})
-    for press, orders in by_press.items():
+    orders = sorted({o for _, o in pairs})
-        try:
+    pbinds = {f"p{i}": p for i, p in enumerate(presses)}
-            obinds: dict[str, Any] = {"plant": plant, "press": press}
+    obinds = {f"o{i}": o for i, o in enumerate(orders)}
-            for i, o in enumerate(orders):
+    pin = ", ".join(f":p{i}" for i in range(len(presses)))
-                obinds[f"o{i}"] = o
     oin = ", ".join(f":o{i}" for i in range(len(orders)))
-
+    try:
-            # 1) cavities + target per active order (article/qty snapshot).
+        # 1) cavities per (press, order) — one query.
-            active = run_select_query(
+        cav_by: dict[tuple[str, str], int] = {
-                "SELECT aoOrdernumber AS o, aoCavities AS cav FROM production.active_orders "
+            (str(r["press"]), str(r["o"])): int(r["cav"] or 0)
-                f"WHERE aoPlant = :plant AND aoPress = :press AND aoOrdernumber IN ({oin})",
+            for r in run_select_query(
-                obinds,
+                "SELECT aoPress AS press, aoOrdernumber AS o, aoCavities AS cav "
+                "FROM production.active_orders "
+                f"WHERE aoPlant = :plant AND aoPress IN ({pin}) AND aoOrdernumber IN ({oin})",
+                {"plant": plant, **pbinds, **obinds},
             )
-            cav_by_order = {str(r["o"]): int(r["cav"] or 0) for r in active}
+        }
-            if not cav_by_order:
+        if not cav_by:
-                continue  # none of these orders are active → notYetSent stays 0
+            return {}  # none of these orders are active → notYetSent stays 0
 
-            # 2) cycle counts grouped by Joborder (one scan for the press).
+        # 2) cycle counts grouped by (Press, Joborder) — one scan. Joborder is unfiltered
-            cyc_groups = [
+        #    (a plain order number OR a JSON array), matched to orders in Python below.
-                (cr["jo"], int(cr["c"] or 0))
+        cyc_by_press: dict[str, list[tuple[Any, int]]] = defaultdict(list)
         for cr in run_select_query(
-                    "SELECT Joborder AS jo, COUNT(*) AS c FROM production.cycles_new "
+            "SELECT Press AS press, Joborder AS jo, COUNT(*) AS c "
-                    "WHERE Press = :press GROUP BY Joborder",
+            f"FROM production.cycles_new WHERE Press IN ({pin}) GROUP BY Press, Joborder",
-                    {"press": press},
+            pbinds,
-                )
-            ]
-
-            # 3) rejects summed per order (one grouped query).
-            rej_by_order: dict[str, int] = {}
-            for rr in run_select_query(
-                "SELECT rejOrdernumber AS o, COALESCE(SUM(rejQuantity),0) AS total "
-                "FROM production.rejects WHERE rejPress = :press "
-                f"AND rejOrdernumber IN ({oin}) GROUP BY rejOrdernumber",
-                obinds,
         ):
-                rej_by_order[str(rr["o"])] = int(rr["total"] or 0)
+            cyc_by_press[str(cr["press"])].append((cr["jo"], int(cr["c"] or 0)))
 
-            def _cycles_for(orderno: str) -> int:
+        # 3) rejects summed per (press, order) — one grouped query.
+        rej_by: dict[tuple[str, str], int] = {
+            (str(rr["press"]), str(rr["o"])): int(rr["total"] or 0)
+            for rr in run_select_query(
+                "SELECT rejPress AS press, rejOrdernumber AS o, "
+                "COALESCE(SUM(rejQuantity),0) AS total FROM production.rejects "
+                f"WHERE rejPress IN ({pin}) AND rejOrdernumber IN ({oin}) "
+                "GROUP BY rejPress, rejOrdernumber",
+                {**pbinds, **obinds},
+            )
+        }
+    except Exception as e:  # noqa: BLE001 — whole map degrades to {} (matches _temper_agg_map)
+        logger.warning("good-parts derivation failed: %s", e)
+        return {}
+
+    def _cycles_for(press: str, orderno: str) -> int:
         total = 0
-                for jo, c in cyc_groups:
+        for jo, c in cyc_by_press.get(press, ()):
             if jo is None:
                 continue
             if jo == orderno:
@@ -230,13 +239,12 @@ def _good_parts_map(plant: str, pairs: set[tuple[str, str]]) -> dict[tuple[str,
                     total += c
         return total
 
-            for orderno, cav in cav_by_order.items():
+    result: dict[tuple[str, str], int] = {}
-                produced = _cycles_for(orderno) * cav
+    for (press, orderno), cav in cav_by.items():
-                good = max(0, produced - rej_by_order.get(orderno, 0))
+        if (press, orderno) not in pairs:   # keep emitted keys identical to the input pairs
-                result[(press, orderno)] = good
-        except Exception as e:  # noqa: BLE001 — degrade this press to no notYetSent
-            logger.warning("good-parts derivation failed for press %s: %s", press, e)
             continue
+        produced = _cycles_for(press, orderno) * cav
+        result[(press, orderno)] = max(0, produced - rej_by.get((press, orderno), 0))
     return result
 
 
@@ -460,7 +468,9 @@ def reasons() -> dict[str, Any]:
     """Active reject reasons — same query as fastpress ``get_reject_reasons``.
     Global (not press-scoped). Read-only."""
     try:
-        rows = run_select_query(
+        # Reference data (admin-edited); served from cache.
+        rows = cached_select(
+            "trj:reject_reasons",
             "SELECT id, reason_de, reason_en FROM settings.reject_reasons "
             "WHERE active = 1 ORDER BY id",
             {},