refactor(eq2db): base read helpers + dead code removal

backend/eq2db/_catalogue.py

@@ -40,6 +40,7 @@
 
 import logging
 import sqlite3
+from collections.abc import Mapping, Sequence
 from pathlib import Path
 from typing import ClassVar
 
@@ -131,6 +132,40 @@ def __enter__(self) -> sqlite3.Connection:
     def __exit__(self, exc_type, exc, tb) -> None:
         self._ctx_conns.pop().close()
 
+    # ── Read helpers ─────────────────────────────────────────────────────────
+
+    def _fetchall(self, sql: str, params: Sequence | Mapping = ()) -> list[sqlite3.Row]:
+        """Run one read query with Row factory; the connection is opened and
+        closed per call. Returns [] when the DB file is missing or the table
+        isn't built yet — eq2db read paths degrade gracefully on an
+        unprovisioned DB rather than 500."""
+        if not self.path.exists():
+            return []
+        conn = sqlite3.connect(self.path)
+        conn.row_factory = sqlite3.Row
+        try:
+            return conn.execute(sql, params).fetchall()
+        except sqlite3.OperationalError:
+            _log.exception("[eq2db] read failed on %r (unbuilt db?)", self)
+            return []
+        finally:
+            conn.close()
+
+    def _fetchone(self, sql: str, params: Sequence | Mapping = ()) -> sqlite3.Row | None:
+        """Single-row variant of :meth:`_fetchall`. None on missing DB,
+        unbuilt table, or no match."""
+        if not self.path.exists():
+            return None
+        conn = sqlite3.connect(self.path)
+        conn.row_factory = sqlite3.Row
+        try:
+            return conn.execute(sql, params).fetchone()
+        except sqlite3.OperationalError:
+            _log.exception("[eq2db] read failed on %r (unbuilt db?)", self)
+            return None
+        finally:
+            conn.close()
+
     # ── DB lifecycle ─────────────────────────────────────────────────────────
 
     def init_db(self) -> sqlite3.Connection:

backend/eq2db/aas.py

@@ -129,20 +129,10 @@ def _create_schema(self, conn: sqlite3.Connection) -> None:
         conn.execute(_SQL["schema_aa_limits"])
         conn.executescript(_SQL["indexes_aas"])
 
-    def _query(self, name: str, params: tuple = (), *, row_factory: bool = False) -> list:
-        """Run one read query; [] when the DB is missing or unbuilt."""
-        if not self.path.exists():
-            return []
-        conn = sqlite3.connect(self.path)
-        if row_factory:
-            conn.row_factory = sqlite3.Row
-        try:
-            return conn.execute(_SQL[name], params).fetchall()
-        except sqlite3.OperationalError:
-            _log.exception("[aas-db] query %s failed (unbuilt db?)", name)
-            return []
-        finally:
-            conn.close()
+    def _query(self, name: str, params: tuple = ()) -> list:
+        """Run one read query by its _SQL block name; [] when the DB is
+        missing or unbuilt (see BaseCatalogue._fetchall)."""
+        return self._fetchall(_SQL[name], params)
 
     def clear_caches(self) -> None:
         """Reset every per-instance cache — used by tests and the build script."""
@@ -196,32 +186,26 @@ def total_max_points(self, tree_types: frozenset[str]) -> int:
         if not tree_types:
             return 0
         if tree_types not in self._total_max_points:
-            if not self.path.exists():
-                return 0
-            conn = sqlite3.connect(self.path)
-            try:
-                placeholders = ",".join("?" * len(tree_types))
-                row = conn.execute(
-                    _SQL["sum_max_points_for_types"].format(placeholders=placeholders),
-                    sorted(tree_types),
-                ).fetchone()
-                self._total_max_points[tree_types] = int(row[0]) if row else 0
-            except sqlite3.OperationalError:
-                return 0
-            finally:
-                conn.close()
+            placeholders = ",".join("?" * len(tree_types))
+            row = self._fetchone(
+                _SQL["sum_max_points_for_types"].format(placeholders=placeholders),
+                sorted(tree_types),
+            )
+            if row is None:
+                return 0  # missing/unbuilt DB — don't cache the zero
+            self._total_max_points[tree_types] = int(row[0]) if row[0] is not None else 0
         return self._total_max_points[tree_types]
 
     def get_tree(self, tree_id: int) -> dict | None:
         """Full tree detail: the aa_trees row plus a ``nodes`` list of aa_nodes
         rows (dicts, DB column names) in tree reading order. None when unknown."""
         if tree_id not in self._trees:
-            trees = self._query("select_tree", (tree_id,), row_factory=True)
+            trees = self._query("select_tree", (tree_id,))
             if not trees:
                 self._trees[tree_id] = None
             else:
                 out = dict(trees[0])
-                nodes = self._query("select_nodes_for_tree", (tree_id,), row_factory=True)
+                nodes = self._query("select_nodes_for_tree", (tree_id,))
                 out["nodes"] = [dict(n) for n in nodes]
                 self._trees[tree_id] = out
         return self._trees[tree_id]

backend/eq2db/raids.py

@@ -86,9 +86,6 @@
 # Column lists
 # ---------------------------------------------------------------------------
 
-_ZONE_SELECT_COLS = _SQL["select_zone_cols"]
-_ENC_SELECT_COLS = _SQL["select_encounter_cols"]
-
 _ACT_TRIGGER_COLS = (
     "id, raid_encounter_id, position, label, notes, "
     "active, regex, sound_data, sound_type, "
@@ -398,105 +395,28 @@ def delete_raid_encounter_by_zone_mob(conn: sqlite3.Connection, *, zone_name: st
 
     # ── Read helpers (path-based) ────────────────────────────────────────────
 
-    def find_zone_by_name(self, name: str) -> dict | None:
-        """Look up a raid zone by name (case-insensitive)."""
-        if not self.path.exists() or not name:
-            return None
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            row = conn.execute(
-                _SQL["find_zone_by_name_ci"].format(cols=_ZONE_SELECT_COLS),
-                (name.lower(),),
-            ).fetchone()
-            return dict(row) if row else None
-
-    def list_encounters_for_zone(self, zone_id: int) -> list[dict]:
-        """All encounters in a raid zone, ordered by position then name."""
-        if not self.path.exists():
-            return []
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            rows = conn.execute(
-                _SQL["list_encounters_for_zone"].format(cols=_ENC_SELECT_COLS),
-                (zone_id,),
-            ).fetchall()
-            return [dict(r) for r in rows]
-
-    def list_zones_by_expansion(self, short: str) -> list[dict]:
-        """All raid zones in an expansion."""
-        if not self.path.exists():
-            return []
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            rows = conn.execute(
-                _SQL["list_zones_by_expansion"].format(cols=_ZONE_SELECT_COLS),
-                (short,),
-            ).fetchall()
-            return [dict(r) for r in rows]
-
     def encounter_revisions(self, encounter_id: int) -> list[dict]:
         """Full revision history for an encounter, newest first."""
-        if not self.path.exists():
-            return []
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            rows = conn.execute(
-                _SQL["list_encounter_revisions"],
-                (encounter_id,),
-            ).fetchall()
-            return [dict(r) for r in rows]
+        return [dict(r) for r in self._fetchall(_SQL["list_encounter_revisions"], (encounter_id,))]
 
     def list_zone_revisions(self, zone_id: int) -> list[dict]:
         """All revision rows for a zone's overview, newest first.
         Each row: {id, edited_at, edited_by, before_md, after_md, edit_note}."""
-        if not self.path.exists():
-            return []
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            rows = conn.execute(
-                _SQL["list_zone_revisions"],
-                (zone_id,),
-            ).fetchall()
-            return [dict(r) for r in rows]
-
-    def stats(self) -> dict:
-        """Diagnostic — counts by table + source."""
-        if not self.path.exists():
-            return {}
-        with sqlite3.connect(self.path) as conn:
-            out = {
-                "zones": conn.execute(_SQL["stats_zones_count"]).fetchone()[0],
-                "encounters": conn.execute(_SQL["stats_encounters_count"]).fetchone()[0],
-                "revisions": conn.execute(_SQL["stats_revisions_count"]).fetchone()[0],
-                "encounters_by_source": dict(conn.execute(_SQL["stats_encounters_by_source"])),
-                "zones_by_expansion": dict(conn.execute(_SQL["stats_zones_by_expansion"])),
-            }
-            return out
+        return [dict(r) for r in self._fetchall(_SQL["list_zone_revisions"], (zone_id,))]
 
     # ── ACT trigger helpers (formerly backend/eq2db/raids_act.py) ────────────
 
     def list_act_triggers_for_encounter(self, encounter_id: int) -> list[dict]:
         """Every ACT trigger row for an encounter, ordered by position then id."""
-        if not self.path.exists():
-            return []
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            rows = conn.execute(
-                f"SELECT {_ACT_TRIGGER_COLS} FROM act_triggers WHERE raid_encounter_id = ? ORDER BY position, id",
-                (encounter_id,),
-            ).fetchall()
-            return [dict(r) for r in rows]
+        rows = self._fetchall(
+            f"SELECT {_ACT_TRIGGER_COLS} FROM act_triggers WHERE raid_encounter_id = ? ORDER BY position, id",
+            (encounter_id,),
+        )
+        return [dict(r) for r in rows]
 
     def get_act_trigger(self, trigger_id: int) -> dict | None:
-        if not self.path.exists():
-            return None
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            row = conn.execute(
-                f"SELECT {_ACT_TRIGGER_COLS} FROM act_triggers WHERE id = ?",
-                (trigger_id,),
-            ).fetchone()
-            return dict(row) if row else None
+        row = self._fetchone(f"SELECT {_ACT_TRIGGER_COLS} FROM act_triggers WHERE id = ?", (trigger_id,))
+        return dict(row) if row else None
 
     @staticmethod
     def upsert_act_trigger(
@@ -574,26 +494,15 @@ def delete_act_trigger(conn: sqlite3.Connection, trigger_id: int) -> bool:
 
     def list_act_spell_timers_for_encounter(self, encounter_id: int) -> list[dict]:
         """Every spell-timer row for an encounter, alphabetical by name."""
-        if not self.path.exists():
-            return []
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            rows = conn.execute(
-                f"SELECT {_ACT_SPELL_TIMER_COLS} FROM act_spell_timers WHERE raid_encounter_id = ? ORDER BY name",
-                (encounter_id,),
-            ).fetchall()
-            return [dict(r) for r in rows]
+        rows = self._fetchall(
+            f"SELECT {_ACT_SPELL_TIMER_COLS} FROM act_spell_timers WHERE raid_encounter_id = ? ORDER BY name",
+            (encounter_id,),
+        )
+        return [dict(r) for r in rows]
 
     def get_act_spell_timer(self, timer_id: int) -> dict | None:
-        if not self.path.exists():
-            return None
-        with sqlite3.connect(self.path) as conn:
-            conn.row_factory = sqlite3.Row
-            row = conn.execute(
-                f"SELECT {_ACT_SPELL_TIMER_COLS} FROM act_spell_timers WHERE id = ?",
-                (timer_id,),
-            ).fetchone()
-            return dict(row) if row else None
+        row = self._fetchone(f"SELECT {_ACT_SPELL_TIMER_COLS} FROM act_spell_timers WHERE id = ?", (timer_id,))
+        return dict(row) if row else None
 
     @staticmethod
     def upsert_act_spell_timer(

backend/eq2db/raids.sql

@@ -298,26 +298,6 @@ VALUES (?, ?, ?, ?, ?, ?);
 -- Read helpers
 -- ---------------------------------------------------------------------------
 
--- Column-list fragments used by find_zone_by_name + list_zones_by_expansion.
--- :name select_zone_cols
-id, zone_name, zone_name_lower, expansion_short, wiki_url,
-access_md, background_md, overview_md,
-level_range, zdiff, lockout_min, lockout_max,
-source, last_synced_at, last_edited_at, last_edited_by
-
--- :name select_encounter_cols
-id, raid_zone_id, mob_name, mob_name_lower, position,
-strategy_md, wiki_url, source, last_synced_at, last_edited_at, last_edited_by
-
--- :name find_zone_by_name_ci
-SELECT {cols} FROM raid_zones WHERE zone_name_lower = ?;
-
--- :name list_encounters_for_zone
-SELECT {cols} FROM raid_encounters WHERE raid_zone_id = ? ORDER BY position, mob_name;
-
--- :name list_zones_by_expansion
-SELECT {cols} FROM raid_zones WHERE expansion_short = ? ORDER BY zone_name;
-
 -- :name list_encounter_revisions
 SELECT id, encounter_id, edited_at, edited_by, before_md, after_md, edit_note
 FROM raid_encounter_revisions
@@ -327,22 +307,3 @@ WHERE encounter_id = ? ORDER BY edited_at DESC, id DESC;
 SELECT id, edited_at, edited_by, before_md, after_md, edit_note
 FROM raid_zone_revisions WHERE raid_zone_id = ?
 ORDER BY edited_at DESC, id DESC;
-
--- ---------------------------------------------------------------------------
--- Stats
--- ---------------------------------------------------------------------------
-
--- :name stats_zones_count
-SELECT COUNT(*) FROM raid_zones;
-
--- :name stats_encounters_count
-SELECT COUNT(*) FROM raid_encounters;
-
--- :name stats_revisions_count
-SELECT COUNT(*) FROM raid_encounter_revisions;
-
--- :name stats_encounters_by_source
-SELECT source, COUNT(*) FROM raid_encounters GROUP BY source;
-
--- :name stats_zones_by_expansion
-SELECT expansion_short, COUNT(*) FROM raid_zones GROUP BY expansion_short ORDER BY 2 DESC;