From cb70cd41afcc3dab7b1d83ffd9b60028e0060e03 Mon Sep 17 00:00:00 2001 From: VortexUK Date: Sat, 11 Jul 2026 10:14:02 +0100 Subject: [PATCH] refactor(eq2db): base _fetchall/_fetchone read helpers + dead code removal Dedup pass over the rearchitected catalogues: - BaseCatalogue gains _fetchall/_fetchone: one place for the missing-file guard, Row factory, per-call connection open/close, and OperationalError graceful degradation (previously only aas._query and zones.list_expansions handled unbuilt DBs; now every read does, and connections are explicitly closed instead of leaking to GC). - 24 copies of the connect/row_factory/exists() boilerplate collapse to helper calls across spells (5 reads), recipes (5), raids (6 incl. the ACT trigger/timer reads), zones (list_expansions, expansion_counts) and aas (_query + total_max_points become thin wrappers). - Dead code removed (zero references anywhere, code/tests/docs): ZoneCatalogue.zone_count, Zone.find_by_id, RaidCatalogue.{find_zone_by_name, list_encounters_for_zone, list_zones_by_expansion, stats}, plus their now-orphaned SQL blocks (raids.sql select_zone_cols/select_encounter_cols/find_zone_by_name_ci/ list_encounters_for_zone/list_zones_by_expansion/stats_*; zones.sql count_zones). - conftest: the seven duplicated DB_PATH + catalogue.path re-point pairs become one loop, so the next module can't forget the catalogue.path half. Behaviour unchanged on live paths: 1521 tests green, pyright clean, character spells / spell-scroll / AA-spell endpoints spot-checked. Co-Authored-By: Claude Fable 5 --- backend/eq2db/_catalogue.py | 35 ++++++++++ backend/eq2db/aas.py | 44 ++++--------- backend/eq2db/raids.py | 123 +++++------------------------------- backend/eq2db/raids.sql | 39 ------------ backend/eq2db/recipes.py | 63 ++++++------------ backend/eq2db/spells.py | 68 +++++++------------- backend/eq2db/zones.py | 46 +++----------- backend/eq2db/zones.sql | 3 - tests/conftest.py | 24 +++---- 9 files changed, 130 insertions(+), 315 deletions(-) diff --git a/backend/eq2db/_catalogue.py b/backend/eq2db/_catalogue.py index 929addc3..84787c73 100644 --- a/backend/eq2db/_catalogue.py +++ b/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: diff --git a/backend/eq2db/aas.py b/backend/eq2db/aas.py index 4c0bdc13..72999958 100644 --- a/backend/eq2db/aas.py +++ b/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] diff --git a/backend/eq2db/raids.py b/backend/eq2db/raids.py index a93af503..f0c3293f 100644 --- a/backend/eq2db/raids.py +++ b/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( diff --git a/backend/eq2db/raids.sql b/backend/eq2db/raids.sql index 528adc96..c8093d9a 100644 --- a/backend/eq2db/raids.sql +++ b/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; diff --git a/backend/eq2db/recipes.py b/backend/eq2db/recipes.py index fa7a051b..a067e39a 100644 --- a/backend/eq2db/recipes.py +++ b/backend/eq2db/recipes.py @@ -288,29 +288,18 @@ def recipe_count(self, conn: sqlite3.Connection) -> int: def find_by_id(self, recipe_id: int) -> RecipeRow | None: """Return a recipe row dict for the given ID, or None.""" - if not self.path.exists(): - return None - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - row = conn.execute(_SQL["find_by_id"].format(cols=_SELECT_COLS), (recipe_id,)).fetchone() + row = self._fetchone(_SQL["find_by_id"].format(cols=_SELECT_COLS), (recipe_id,)) return _row_to_dict(row) if row else None def find_by_name(self, name: str) -> list[RecipeRow]: """Return recipes whose name matches (exact then LIKE), ordered by name.""" - if not self.path.exists(): - return [] - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - rows = conn.execute( - _SQL["find_by_name_exact"].format(cols=_SELECT_COLS), - (name.lower(),), - ).fetchall() - if not rows: - # LIKE fallback — escape user wildcards (BE-006). - rows = conn.execute( - _SQL["find_by_name_like"].format(cols=_SELECT_COLS), - (f"%{like_escape(name.lower())}%",), - ).fetchall() + rows = self._fetchall(_SQL["find_by_name_exact"].format(cols=_SELECT_COLS), (name.lower(),)) + if not rows: + # LIKE fallback — escape user wildcards (BE-006). + rows = self._fetchall( + _SQL["find_by_name_like"].format(cols=_SELECT_COLS), + (f"%{like_escape(name.lower())}%",), + ) return [_row_to_dict(r) for r in rows] def find_by_spell(self, spell_name: str, tier: str) -> list[RecipeRow]: @@ -321,14 +310,10 @@ def find_by_spell(self, spell_name: str, tier: str) -> list[RecipeRow]: Matched case-insensitively against base_name_lower. tier: One of the SPELL_TIERS values, e.g. "Expert". """ - if not self.path.exists(): - return [] - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - rows = conn.execute( - _SQL["find_by_spell"].format(cols=_SELECT_COLS), - (spell_name.lower(), tier), - ).fetchall() + rows = self._fetchall( + _SQL["find_by_spell"].format(cols=_SELECT_COLS), + (spell_name.lower(), tier), + ) return [_row_to_dict(r) for r in rows] def find_spells_by_tier(self, spell_names: list[str], tier: str) -> dict[str, RecipeRow]: @@ -338,28 +323,22 @@ def find_spells_by_tier(self, spell_names: list[str], tier: str) -> dict[str, Re Recipes not found in the DB are omitted from the result. Designed for the spellcheck upgrade-path feature (one DB query for N spells). """ - if not self.path.exists() or not spell_names: + if not spell_names: return {} placeholders = ",".join("?" * len(spell_names)) params = [n.lower() for n in spell_names] + [tier] - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - rows = conn.execute( - _SQL["find_spells_by_tier"].format(cols=_SELECT_COLS, placeholders=placeholders), - params, - ).fetchall() + rows = self._fetchall( + _SQL["find_spells_by_tier"].format(cols=_SELECT_COLS, placeholders=placeholders), + params, + ) return {r["base_name_lower"]: _row_to_dict(r) for r in rows} def find_by_output_id(self, item_id: int) -> list[RecipeRow]: """Return all recipes that produce the given item ID at any quality tier.""" - if not self.path.exists(): - return [] - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - rows = conn.execute( - _SQL["find_by_output_id"].format(cols=_SELECT_COLS), - {"id": item_id}, - ).fetchall() + rows = self._fetchall( + _SQL["find_by_output_id"].format(cols=_SELECT_COLS), + {"id": item_id}, + ) return [_row_to_dict(r) for r in rows] diff --git a/backend/eq2db/spells.py b/backend/eq2db/spells.py index 58589e5d..aa76c00d 100644 --- a/backend/eq2db/spells.py +++ b/backend/eq2db/spells.py @@ -363,24 +363,18 @@ def spell_count(self, conn: sqlite3.Connection) -> int: def find_by_id(self, spell_id: int) -> SpellRow | None: """Return a spell row dict for the given ID, or None.""" - if not self.path.exists(): - return None - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - row = conn.execute(_SQL["find_by_id"].format(cols=_SELECT_COLS), (spell_id,)).fetchone() + row = self._fetchone(_SQL["find_by_id"].format(cols=_SELECT_COLS), (spell_id,)) return _row_to_dict(row) if row else None def find_by_ids(self, spell_ids: list[int]) -> dict[int, SpellRow]: """Return {spell_id: row_dict} for all matching IDs. Missing IDs are omitted.""" - if not spell_ids or not self.path.exists(): + if not spell_ids: return {} placeholders = ",".join("?" * len(spell_ids)) - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - rows = conn.execute( - _SQL["find_by_ids"].format(cols=_SELECT_COLS, placeholders=placeholders), - spell_ids, - ).fetchall() + rows = self._fetchall( + _SQL["find_by_ids"].format(cols=_SELECT_COLS, placeholders=placeholders), + spell_ids, + ) return {row["id"]: _row_to_dict(row) for row in rows} def upgradeable_crcs(self, crcs: Iterable[int | None]) -> set[int]: @@ -396,11 +390,10 @@ def upgradeable_crcs(self, crcs: Iterable[int | None]) -> set[int]: Empty input (or a missing DB) → empty set. """ ids = [c for c in {*crcs} if c is not None] - if not ids or not self.path.exists(): + if not ids: return set() placeholders = ",".join("?" * len(ids)) - with sqlite3.connect(self.path) as conn: - rows = conn.execute(_SQL["upgradeable_crcs"].format(placeholders=placeholders), ids).fetchall() + rows = self._fetchall(_SQL["upgradeable_crcs"].format(placeholders=placeholders), ids) return {r[0] for r in rows} def find_by_crc(self, crc: int, tier: int | None = None) -> SpellRow | None: @@ -414,42 +407,25 @@ def find_by_crc(self, crc: int, tier: int | None = None) -> SpellRow | None: key = (crc, tier) if key in self._crc_cache: return self._crc_cache[key] - result: SpellRow | None = None - if self.path.exists(): - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - row = None - if tier is not None: - row = conn.execute( - _SQL["find_by_crc_and_tier"].format(cols=_SELECT_COLS), - (crc, tier), - ).fetchone() - if row is None: - # Fallback: highest available tier - row = conn.execute( - _SQL["find_by_crc_highest_tier"].format(cols=_SELECT_COLS), - (crc,), - ).fetchone() - result = _row_to_dict(row) if row else None + row = None + if tier is not None: + row = self._fetchone(_SQL["find_by_crc_and_tier"].format(cols=_SELECT_COLS), (crc, tier)) + if row is None: + # Fallback: highest available tier + row = self._fetchone(_SQL["find_by_crc_highest_tier"].format(cols=_SELECT_COLS), (crc,)) + result = _row_to_dict(row) if row else None self._crc_cache[key] = result return result def find_by_name(self, name: str) -> list[SpellRow]: """Return all spell rows whose name matches (exact, then LIKE). Ordered by level.""" - if not self.path.exists(): - return [] - with sqlite3.connect(self.path) as conn: - conn.row_factory = sqlite3.Row - rows = conn.execute( - _SQL["find_by_name_exact"].format(cols=_SELECT_COLS), - (name.lower(),), - ).fetchall() - if not rows: - # LIKE fallback — escape user wildcards (BE-006). - rows = conn.execute( - _SQL["find_by_name_like"].format(cols=_SELECT_COLS), - (f"%{like_escape(name.lower())}%",), - ).fetchall() + rows = self._fetchall(_SQL["find_by_name_exact"].format(cols=_SELECT_COLS), (name.lower(),)) + if not rows: + # LIKE fallback — escape user wildcards (BE-006). + rows = self._fetchall( + _SQL["find_by_name_like"].format(cols=_SELECT_COLS), + (f"%{like_escape(name.lower())}%",), + ) return [_row_to_dict(r) for r in rows] def character_upgradeable_spells(self, spell_ids: list[int]) -> list[SpellRow]: diff --git a/backend/eq2db/zones.py b/backend/eq2db/zones.py index 86159753..ea163883 100644 --- a/backend/eq2db/zones.py +++ b/backend/eq2db/zones.py @@ -179,19 +179,6 @@ def to_dict(self) -> dict: # Lookups # ----------------------------------------------------------------- - @classmethod - def find_by_id(cls, zone_id: int, path: Path = DB_PATH) -> Zone | None: - """Single-zone fetch by primary key. None if not found.""" - if not path.exists(): - return None - with sqlite3.connect(path) as conn: - conn.row_factory = sqlite3.Row - row = conn.execute( - _SQL["find_zone_by_id"].format(cols=_SELECT_COLS), - (zone_id,), - ).fetchone() - return cls._from_row(conn, row) if row else None - @classmethod def find_by_name(cls, name: str, path: Path = DB_PATH) -> Zone | None: """Resolve a zone by name, falling back to the alias table. @@ -1348,9 +1335,6 @@ def upsert_zones(self, zones: list[dict], conn: sqlite3.Connection) -> int: n += 1 return n - def zone_count(self, conn: sqlite3.Connection) -> int: - return conn.execute(_SQL["count_zones"]).fetchone()[0] - def replace_bosses_for_zone(self, conn: sqlite3.Connection, zone_id: int, encounters: list[dict]) -> int: """Replace the encounters list for a zone. Atomic per-zone. Delegates to ``ZoneEncounter.replace_all_for_zone`` (takes an open @@ -1395,29 +1379,19 @@ def list_expansions(self) -> list[dict]: Returns [] when zones.db is missing or the zones table does not yet exist (graceful degradation — the admin endpoint must never 500 on a missing DB). """ - if not self.path.exists(): - return [] - try: - with sqlite3.connect(self.path) as conn: - rows = conn.execute(_SQL["list_distinct_expansions"]).fetchall() - # De-duplicate by short (same short can have multiple rows with the same year). - seen: set[str] = set() - result: list[dict] = [] - for short, name, _year in rows: - if short not in seen: - seen.add(short) - result.append({"short": short, "name": name}) - return result - except sqlite3.OperationalError: - # zones table may not exist yet (e.g. pre-seeded zones.db stub). - return [] + rows = self._fetchall(_SQL["list_distinct_expansions"]) + # De-duplicate by short (same short can have multiple rows with the same year). + seen: set[str] = set() + result: list[dict] = [] + for short, name, _year in rows: + if short not in seen: + seen.add(short) + result.append({"short": short, "name": name}) + return result def expansion_counts(self) -> dict[str, int]: """Diagnostic: zones per expansion short. Used by the build report.""" - if not self.path.exists(): - return {} - with sqlite3.connect(self.path) as conn: - return dict(conn.execute(_SQL["expansion_counts"])) + return {r[0]: r[1] for r in self._fetchall(_SQL["expansion_counts"])} # ── Editable encounter + mob CRUD (used by the zones-admin routes) ─────── diff --git a/backend/eq2db/zones.sql b/backend/eq2db/zones.sql index c9e06df2..51c3687e 100644 --- a/backend/eq2db/zones.sql +++ b/backend/eq2db/zones.sql @@ -220,9 +220,6 @@ SELECT name, expansion_short FROM zones WHERE id = ?; -- :name select_zone_name_by_id SELECT name FROM zones WHERE id = ?; --- :name count_zones -SELECT COUNT(*) FROM zones; - -- :name find_zone_by_name_lower SELECT {cols} FROM zones WHERE name_lower = ? LIMIT 1; diff --git a/tests/conftest.py b/tests/conftest.py index 013e1ec4..f88094d7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -114,18 +114,18 @@ def pytest_configure(config: pytest.Config) -> None: # noqa: ARG001 parses_db.DB_PATH = resolve_db_path("DB_PARSES_PATH", "parses", "parses.db") users_db.DB_PATH = resolve_db_path("DB_USERS_PATH", "users.db") census_store.DB_PATH = resolve_db_path("DB_CENSUS_PATH", "census", "census.db") - zones_db.DB_PATH = resolve_db_path("DB_ZONES_PATH", "zones", "zones.db") - zones_db.catalogue.path = zones_db.DB_PATH - spells_db.DB_PATH = resolve_db_path("DB_SPELLS_PATH", "spells", "spells.db") - spells_db.catalogue.path = spells_db.DB_PATH - recipes_db.DB_PATH = resolve_db_path("DB_RECIPES_PATH", "recipes", "recipes.db") - recipes_db.catalogue.path = recipes_db.DB_PATH - raids_db.DB_PATH = resolve_db_path("DB_RAIDS_PATH", "raids", "raids.db") - raids_db.catalogue.path = raids_db.DB_PATH - items_db.DB_PATH = resolve_db_path("DB_ITEMS_PATH", "items", "items.db") - items_db.catalogue.path = items_db.DB_PATH - classes_db.DB_PATH = resolve_db_path("DB_CLASSES_PATH", "classes", "classes.db") - classes_db.catalogue.path = classes_db.DB_PATH + # eq2db catalogue modules: re-point both the module constant AND the + # shared catalogue instance (its path was captured at import time). + for mod, env_var, subdir, filename in ( + (zones_db, "DB_ZONES_PATH", "zones", "zones.db"), + (spells_db, "DB_SPELLS_PATH", "spells", "spells.db"), + (recipes_db, "DB_RECIPES_PATH", "recipes", "recipes.db"), + (raids_db, "DB_RAIDS_PATH", "raids", "raids.db"), + (items_db, "DB_ITEMS_PATH", "items", "items.db"), + (classes_db, "DB_CLASSES_PATH", "classes", "classes.db"), + ): + mod.DB_PATH = resolve_db_path(env_var, subdir, filename) + mod.catalogue.path = mod.DB_PATH # Create both schemas immediately. FastAPI's startup hooks (which would # normally call init_db) don't fire under ASGITransport, so without this