From 979aae426bbe9c40e1e0cde30c19ec91d670faac Mon Sep 17 00:00:00 2001 From: Erik Date: Thu, 9 Jul 2026 12:06:49 +0200 Subject: [PATCH] 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) --- _db.py | 4 +- api/routes.py | 140 +++++++++++++++++++++++++++----------------------- 2 files changed, 77 insertions(+), 67 deletions(-) diff --git a/_db.py b/_db.py index 57cdac3..e5f3181 100644 --- a/_db.py +++ b/_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 diff --git a/api/routes.py b/api/routes.py index a624072..2df5811 100644 --- a/api/routes.py +++ b/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,71 +174,77 @@ 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) - for p, o in pairs: - by_press[p].append(o) + # Batched across ALL presses: 3 grouped queries total (was 3 per press = N+1). + # Semantics preserved from the per-press version — only the query scope changed + # (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. + presses = sorted({p for p, _ in pairs}) + orders = sorted({o for _, o in pairs}) + pbinds = {f"p{i}": p for i, p in enumerate(presses)} + obinds = {f"o{i}": o for i, o in enumerate(orders)} + pin = ", ".join(f":p{i}" for i in range(len(presses))) + oin = ", ".join(f":o{i}" for i in range(len(orders))) + try: + # 1) cavities per (press, order) — one query. + cav_by: dict[tuple[str, str], int] = { + (str(r["press"]), str(r["o"])): int(r["cav"] or 0) + for r in run_select_query( + "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}, + ) + } + if not cav_by: + return {} # none of these orders are active → notYetSent stays 0 + + # 2) cycle counts grouped by (Press, Joborder) — one scan. Joborder is unfiltered + # (a plain order number OR a JSON array), matched to orders in Python below. + cyc_by_press: dict[str, list[tuple[Any, int]]] = defaultdict(list) + for cr in run_select_query( + "SELECT Press AS press, Joborder AS jo, COUNT(*) AS c " + f"FROM production.cycles_new WHERE Press IN ({pin}) GROUP BY Press, Joborder", + pbinds, + ): + cyc_by_press[str(cr["press"])].append((cr["jo"], int(cr["c"] or 0))) + + # 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_by_press.get(press, ()): + if jo is None: + continue + if jo == orderno: + total += c + elif isinstance(jo, str) and orderno in jo: + try: + arr = json.loads(jo) + except Exception: # noqa: BLE001 + arr = None + if isinstance(arr, list) and orderno in arr: + total += c + return total result: dict[tuple[str, str], int] = {} - for press, orders in by_press.items(): - try: - obinds: dict[str, Any] = {"plant": plant, "press": press} - for i, o in enumerate(orders): - obinds[f"o{i}"] = o - oin = ", ".join(f":o{i}" for i in range(len(orders))) - - # 1) cavities + target per active order (article/qty snapshot). - active = run_select_query( - "SELECT aoOrdernumber AS o, aoCavities AS cav FROM production.active_orders " - f"WHERE aoPlant = :plant AND aoPress = :press AND aoOrdernumber IN ({oin})", - obinds, - ) - cav_by_order = {str(r["o"]): int(r["cav"] or 0) for r in active} - if not cav_by_order: - continue # none of these orders are active → notYetSent stays 0 - - # 2) cycle counts grouped by Joborder (one scan for the press). - cyc_groups = [ - (cr["jo"], int(cr["c"] or 0)) - for cr in run_select_query( - "SELECT Joborder AS jo, COUNT(*) AS c FROM production.cycles_new " - "WHERE Press = :press GROUP BY Joborder", - {"press": press}, - ) - ] - - # 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) - - def _cycles_for(orderno: str) -> int: - total = 0 - for jo, c in cyc_groups: - if jo is None: - continue - if jo == orderno: - total += c - elif isinstance(jo, str) and orderno in jo: - try: - arr = json.loads(jo) - except Exception: # noqa: BLE001 - arr = None - if isinstance(arr, list) and orderno in arr: - total += c - return total - - for orderno, cav in cav_by_order.items(): - produced = _cycles_for(orderno) * cav - good = max(0, produced - rej_by_order.get(orderno, 0)) - 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) + for (press, orderno), cav in cav_by.items(): + if (press, orderno) not in pairs: # keep emitted keys identical to the input pairs 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", {},