refactor(census): promote BaseCatalogue + convert store.py to CensusStore

CLAUDE.md

@@ -12,7 +12,7 @@ A web companion site for EverQuest 2 (TLE), with a Discord bot for spot checks (
 | `backend/census/models.py` | Dataclasses: `ItemData`, `ItemStat`, `ItemEffect`, `GuildData`, `GuildMember`, `CharacterSpells`, `SpellEntry`, `NodeAA`, `CharacterAAs` |
 | `backend/census/constants.py` | `STAT_MAP` (stat display names/groups), class frozensets (`FIGHTERS`, `PRIESTS`, `SCOUTS`, `MAGES`, `ARTISANS`), `ARCHETYPES`, `CLASS_GROUPS`, `TYPEINFO_DISPLAY`, `ITEM_DISPLAY` |
 | `backend/census/item_parser.py` | Item data parsing helpers (parse_item, parse_stats, parse_effects, parse_flags, etc.) extracted from client.py |
-| `backend/eq2db/spells.py` | Local SQLite spell catalogue — `SpellCatalogue` class (shared `catalogue` instance): strip_roman, unique_highest_entries, load_blocklist, find_by_ids, find_by_crc, spell_to_row, upsert_spells. Every eq2db module follows this convention: pure helpers are staticmethods with full bodies in the class, DB reads are instance methods on a per-instance `path`, tests construct `XCatalogue(tmp_db)` and patch the instance, `tests/conftest.py` re-points `catalogue.path`. All seven catalogues inherit `BaseCatalogue` (`backend/eq2db/_catalogue.py`) — `init_db` is a template method (connection preamble + commit in the base); subclasses implement `_create_schema` (+ optional `_post_init` backfills), set `FOREIGN_KEYS`/`CREATE_META` class flags, and override `clear_caches` when they hold caches. |
+| `backend/eq2db/spells.py` | Local SQLite spell catalogue — `SpellCatalogue` class (shared `catalogue` instance): strip_roman, unique_highest_entries, load_blocklist, find_by_ids, find_by_crc, spell_to_row, upsert_spells. Every eq2db module follows this convention: pure helpers are staticmethods with full bodies in the class, DB reads are instance methods on a per-instance `path`, tests construct `XCatalogue(tmp_db)` and patch the instance, `tests/conftest.py` re-points `catalogue.path`. All seven catalogues inherit `BaseCatalogue` (`backend/db_catalogue.py`) — `init_db` is a template method (connection preamble + commit in the base); subclasses implement `_create_schema` (+ optional `_post_init` backfills), set `FOREIGN_KEYS`/`CREATE_META` class flags, and override `clear_caches` when they hold caches. |
 | `backend/eq2db/recipes.py` | Local SQLite recipe catalogue (~70k rows) — `RecipeCatalogue`: find_by_id, find_by_name, find_by_spell, find_spells_by_tier, find_by_output_id. Secondary components stored as JSON array. |
 | `backend/eq2db/aas.py` | Local SQLite AA catalogue (aas.db — committed like classes.db). Tables `aa_trees` (tree_type + max_points precomputed at build) + `aa_nodes` + `aa_limits` (per-xpac caps from aa_limits.json). Accessors: `load_tree_index`, `tree_node_costs`, `tree_max_points`, `total_max_points`, `get_tree`, `xpac_limits` (short-code alias tolerant); `detect_tree_type` is build-time only. Rebuild: `scripts/download_aa_trees.py` (tree JSONs are local intermediates, gitignored) then `scripts/build_aas_db.py`; commit the db. |
 | `backend/eq2db/zones.py` | Local SQLite zone catalogue (~1124 rows) — `ZoneCatalogue` over frozen dataclass models (`Zone`, `ZoneEncounter`, `ZoneEncounterMob`, `FeaturedRaid*`). Tables: `zones`, `zone_types`, `zone_aliases`, `zone_encounters`, `zone_encounter_mobs` + the featured-raid trio. Lookups: `find_by_name`, `list_by_expansion`, `list_by_event`, `list_by_type`, `list_bosses_for_zone`, `find_zones_by_boss`. Zone metadata from `scripts/dev/eq2_zones.cleaned.json`; rebuild via `scripts/build_zones_db.py`. Boss data is curator-managed in-place. |

backend/census/store.py

@@ -3,7 +3,11 @@
 never blocks on Census; background refreshes merge in fresh data "keep best
 known" — a sparse Census response never nulls out good data.
 
-Mirrors parses/db.py: DB_CENSUS_PATH env override, WAL, idempotent _MIGRATIONS.
+All behaviour lives on :class:`CensusStore` (the catalogue convention — see
+backend/db_catalogue.py): the shared module-level ``store`` instance is the
+runtime entry point (consumers alias it ``census_store``); the get/upsert
+helpers take an open conn (callers batch reads/writes per connection) and are
+staticmethods. Tests construct ``CensusStore(tmp_db)``.
 """
 
 from __future__ import annotations
@@ -15,6 +19,7 @@
 from pathlib import Path
 from typing import Any, TypedDict
 
+from backend.db_catalogue import BaseCatalogue
 from backend.db_helpers import resolve_db_path
 
 
@@ -74,147 +79,170 @@ class StoreRecord(TypedDict):
 _MIGRATIONS: list[str] = []  # future schema bumps appended here
 
 
-def init_db(path: Path = DB_PATH) -> sqlite3.Connection:
-    """Create tables if missing. Returns an open connection."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    conn = sqlite3.connect(path)
-    conn.execute("PRAGMA journal_mode = WAL;")
-    conn.execute("PRAGMA synchronous = NORMAL;")
-    conn.execute("PRAGMA foreign_keys = ON;")
-    conn.execute(_CREATE_CHARACTERS)
-    conn.execute(_CREATE_GUILDS)
-    conn.execute(_CREATE_CHARACTER_AAS)
-    for stmt in _MIGRATIONS:
-        try:
-            conn.execute(stmt)
-        except sqlite3.OperationalError as exc:
-            _log.info(
-                "[census-store] migration skipped (likely already applied): %s — %s",
-                stmt,
-                exc,
-            )
-    conn.commit()
-    return conn
-
-
-def upsert_character(
-    conn: sqlite3.Connection,
-    name: str,
-    world: str,
-    data: dict,
-    *,
-    resolved: bool,
-    now: int | None = None,
-) -> None:
-    """Merge-store a character (keep best-known).
-
-    When ``resolved`` is False the call is a no-op (never overwrite a good row
-    with a sparse one, never insert a sparse first-sight row).
-
-    When True and a row already exists, the incoming blob is **overlaid** onto
-    the stored one field-by-field: keys present in ``data`` refresh the stored
-    values, keys ``data`` omits are preserved. This is what stops a guild-roster
-    overview (name/level/class/deity only — no equipment/stats) from nulling an
-    individually-resolved character's gear on the next roster refresh. A partial
-    write (one that carries no ``equipment`` key, i.e. hasn't re-resolved the
-    full profile) also leaves ``last_resolved_at`` untouched so the character
-    view's staleness clock stays honest and still triggers a real refresh."""
-    if not resolved:
-        return
-    write_ts = int(time.time()) if now is None else now
-    resolved_ts = write_ts
-
-    existing = get_character(conn, name, world)
-    if existing is not None:
-        incoming_is_full = "equipment" in data
-        data = {**existing["data"], **data}
-        if not incoming_is_full:
-            resolved_ts = existing["last_resolved_at"]  # sparse overlay — don't advance freshness
-
-    conn.execute(
-        """
-        INSERT INTO characters (name_lower, world, name, level, guild_name, data_json, last_resolved_at, updated_at)
-        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-        ON CONFLICT(name_lower, world) DO UPDATE SET
-            name=excluded.name, level=excluded.level, guild_name=excluded.guild_name,
-            data_json=excluded.data_json, last_resolved_at=excluded.last_resolved_at,
-            updated_at=excluded.updated_at
-        """,
-        (name.lower(), world, name, data.get("level"), data.get("guild_name"), json.dumps(data), resolved_ts, write_ts),
-    )
-    conn.commit()
-
-
-def get_character(conn: sqlite3.Connection, name: str, world: str) -> StoreRecord | None:
-    """Return {data, last_resolved_at} or None."""
-    row = conn.execute(
-        "SELECT data_json, last_resolved_at FROM characters WHERE name_lower=? AND world=?",
-        (name.lower(), world),
-    ).fetchone()
-    if row is None:
-        return None
-    return {"data": json.loads(row[0]), "last_resolved_at": row[1]}
-
-
-def upsert_guild(conn: sqlite3.Connection, name: str, world: str, data: dict, *, now: int | None = None) -> None:
-    """Store the guild roster blob (member names+ranks + info). Always replaces —
-    the roster list is reliable from Census regardless of member login recency."""
-    ts = int(time.time()) if now is None else now
-    conn.execute(
-        """
-        INSERT INTO guilds (name_lower, world, name, data_json, last_resolved_at, updated_at)
-        VALUES (?, ?, ?, ?, ?, ?)
-        ON CONFLICT(name_lower, world) DO UPDATE SET
-            name=excluded.name, data_json=excluded.data_json,
-            last_resolved_at=excluded.last_resolved_at, updated_at=excluded.updated_at
-        """,
-        (name.lower(), world, name, json.dumps(data), ts, ts),
-    )
-    conn.commit()
-
-
-def get_guild(conn: sqlite3.Connection, name: str, world: str) -> StoreRecord | None:
-    row = conn.execute(
-        "SELECT data_json, last_resolved_at FROM guilds WHERE name_lower=? AND world=?",
-        (name.lower(), world),
-    ).fetchone()
-    if row is None:
-        return None
-    return {"data": json.loads(row[0]), "last_resolved_at": row[1]}
-
-
-def get_character_aas(conn: sqlite3.Connection, name: str, world: str) -> StoreRecord | None:
-    """Return the persisted CharAAsResponse dict (or None) for (name, world).
-
-    The record carries the model_dump() of the response plus a
-    last_resolved_at unix timestamp."""
-    row = conn.execute(
-        "SELECT data_json, last_resolved_at FROM character_aas WHERE name_lower = ? AND world = ?",
-        (name.lower(), world),
-    ).fetchone()
-    if row is None:
-        return None
-    return {
-        "data": json.loads(row[0]),
-        "last_resolved_at": row[1],
-    }
-
-
-def upsert_character_aas(
-    conn: sqlite3.Connection,
-    name: str,
-    world: str,
-    data: dict,
-    *,
-    now: int | None = None,
-) -> None:
-    """Insert or update the (name, world) AA record. Always overwrites — AAs
-    have no 'best-known merge' equivalent because the Census response is
-    authoritative."""
-    if now is None:
-        now = int(time.time())
-    conn.execute(
-        "INSERT OR REPLACE INTO character_aas (name_lower, world, data_json, last_resolved_at) VALUES (?, ?, ?, ?)",
-        (name.lower(), world, json.dumps(data), now),
-    )
-    conn.commit()
+class CensusStore(BaseCatalogue):
+    """Read/write access to one census.db file (last-known Census lookups)."""
+
+    FOREIGN_KEYS = True
+
+    # census.db predates the shared _meta table and has no build provenance
+    # to track — rows carry their own timestamps.
+    CREATE_META = False
+
+    def __init__(self, path: Path = DB_PATH) -> None:
+        super().__init__(path)
+
+    def _create_schema(self, conn: sqlite3.Connection) -> None:
+        conn.execute(_CREATE_CHARACTERS)
+        conn.execute(_CREATE_GUILDS)
+        conn.execute(_CREATE_CHARACTER_AAS)
+        for stmt in _MIGRATIONS:
+            try:
+                conn.execute(stmt)
+            except sqlite3.OperationalError as exc:
+                _log.info(
+                    "[census-store] migration skipped (likely already applied): %s — %s",
+                    stmt,
+                    exc,
+                )
+
+    # ── Characters ───────────────────────────────────────────────────────────
+
+    @staticmethod
+    def upsert_character(
+        conn: sqlite3.Connection,
+        name: str,
+        world: str,
+        data: dict,
+        *,
+        resolved: bool,
+        now: int | None = None,
+    ) -> None:
+        """Merge-store a character (keep best-known).
+
+        When ``resolved`` is False the call is a no-op (never overwrite a good row
+        with a sparse one, never insert a sparse first-sight row).
+
+        When True and a row already exists, the incoming blob is **overlaid** onto
+        the stored one field-by-field: keys present in ``data`` refresh the stored
+        values, keys ``data`` omits are preserved. This is what stops a guild-roster
+        overview (name/level/class/deity only — no equipment/stats) from nulling an
+        individually-resolved character's gear on the next roster refresh. A partial
+        write (one that carries no ``equipment`` key, i.e. hasn't re-resolved the
+        full profile) also leaves ``last_resolved_at`` untouched so the character
+        view's staleness clock stays honest and still triggers a real refresh."""
+        if not resolved:
+            return
+        write_ts = int(time.time()) if now is None else now
+        resolved_ts = write_ts
+
+        existing = CensusStore.get_character(conn, name, world)
+        if existing is not None:
+            incoming_is_full = "equipment" in data
+            data = {**existing["data"], **data}
+            if not incoming_is_full:
+                resolved_ts = existing["last_resolved_at"]  # sparse overlay — don't advance freshness
+
+        conn.execute(
+            """
+            INSERT INTO characters (name_lower, world, name, level, guild_name, data_json, last_resolved_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(name_lower, world) DO UPDATE SET
+                name=excluded.name, level=excluded.level, guild_name=excluded.guild_name,
+                data_json=excluded.data_json, last_resolved_at=excluded.last_resolved_at,
+                updated_at=excluded.updated_at
+            """,
+            (
+                name.lower(),
+                world,
+                name,
+                data.get("level"),
+                data.get("guild_name"),
+                json.dumps(data),
+                resolved_ts,
+                write_ts,
+            ),
+        )
+        conn.commit()
+
+    @staticmethod
+    def get_character(conn: sqlite3.Connection, name: str, world: str) -> StoreRecord | None:
+        """Return {data, last_resolved_at} or None."""
+        row = conn.execute(
+            "SELECT data_json, last_resolved_at FROM characters WHERE name_lower=? AND world=?",
+            (name.lower(), world),
+        ).fetchone()
+        if row is None:
+            return None
+        return {"data": json.loads(row[0]), "last_resolved_at": row[1]}
+
+    # ── Guilds ───────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def upsert_guild(conn: sqlite3.Connection, name: str, world: str, data: dict, *, now: int | None = None) -> None:
+        """Store the guild roster blob (member names+ranks + info). Always replaces —
+        the roster list is reliable from Census regardless of member login recency."""
+        ts = int(time.time()) if now is None else now
+        conn.execute(
+            """
+            INSERT INTO guilds (name_lower, world, name, data_json, last_resolved_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT(name_lower, world) DO UPDATE SET
+                name=excluded.name, data_json=excluded.data_json,
+                last_resolved_at=excluded.last_resolved_at, updated_at=excluded.updated_at
+            """,
+            (name.lower(), world, name, json.dumps(data), ts, ts),
+        )
+        conn.commit()
+
+    @staticmethod
+    def get_guild(conn: sqlite3.Connection, name: str, world: str) -> StoreRecord | None:
+        row = conn.execute(
+            "SELECT data_json, last_resolved_at FROM guilds WHERE name_lower=? AND world=?",
+            (name.lower(), world),
+        ).fetchone()
+        if row is None:
+            return None
+        return {"data": json.loads(row[0]), "last_resolved_at": row[1]}
+
+    # ── Character AAs ────────────────────────────────────────────────────────
+
+    @staticmethod
+    def get_character_aas(conn: sqlite3.Connection, name: str, world: str) -> StoreRecord | None:
+        """Return the persisted CharAAsResponse dict (or None) for (name, world).
+
+        The record carries the model_dump() of the response plus a
+        last_resolved_at unix timestamp."""
+        row = conn.execute(
+            "SELECT data_json, last_resolved_at FROM character_aas WHERE name_lower = ? AND world = ?",
+            (name.lower(), world),
+        ).fetchone()
+        if row is None:
+            return None
+        return {
+            "data": json.loads(row[0]),
+            "last_resolved_at": row[1],
+        }
+
+    @staticmethod
+    def upsert_character_aas(
+        conn: sqlite3.Connection,
+        name: str,
+        world: str,
+        data: dict,
+        *,
+        now: int | None = None,
+    ) -> None:
+        """Insert or update the (name, world) AA record. Always overwrites — AAs
+        have no 'best-known merge' equivalent because the Census response is
+        authoritative."""
+        if now is None:
+            now = int(time.time())
+        conn.execute(
+            "INSERT OR REPLACE INTO character_aas (name_lower, world, data_json, last_resolved_at) VALUES (?, ?, ?, ?)",
+            (name.lower(), world, json.dumps(data), now),
+        )
+        conn.commit()
+
+
+# The shared default instance — every runtime consumer goes through this.
+store = CensusStore()

backend/eq2db/_catalogue.py → backend/db_catalogue.py

@@ -1,6 +1,9 @@
-"""Shared base class for the eq2db catalogue classes.
+"""Shared base class for SQLite catalogue/store classes.
 
-Every data module under ``backend/eq2db/`` exposes one catalogue class
+Originally extracted from the eq2db data modules; now the base for every
+per-file SQLite data interface in the codebase (eq2db catalogues, the
+census store, ...). Every data module under ``backend/eq2db/`` exposes one
+catalogue class
 (AACatalogue, ClassCatalogue, SpellCatalogue, RecipeCatalogue,
 ZoneCatalogue, ItemCatalogue, RaidCatalogue) following the same
 convention:

backend/eq2db/aas.py

@@ -25,9 +25,9 @@
 from pathlib import Path
 from typing import Any
 
+from backend.db_catalogue import BaseCatalogue
 from backend.db_helpers import resolve_db_path
 from backend.eq2db import _meta as _meta_db
-from backend.eq2db._catalogue import BaseCatalogue
 from backend.sql_loader import load_sql
 
 _log = logging.getLogger(__name__)

backend/eq2db/classes.py

@@ -36,8 +36,8 @@
 from pathlib import Path
 from typing import Any, TypeVar
 
+from backend.db_catalogue import BaseCatalogue
 from backend.db_helpers import resolve_db_path
-from backend.eq2db._catalogue import BaseCatalogue
 from backend.sql_loader import load_sql
 
 _T = TypeVar("_T")

backend/eq2db/items.py

@@ -25,8 +25,8 @@
 from backend.census._coerce import coerce_int as _coerce_int
 from backend.census._coerce import coerce_str_or_none as _coerce_str
 from backend.census.item_level import compute_ilvl
+from backend.db_catalogue import BaseCatalogue
 from backend.db_helpers import like_escape, resolve_db_path
-from backend.eq2db._catalogue import BaseCatalogue
 from backend.eq2db.classes import catalogue as _classes
 from backend.sql_loader import load_sql
 

backend/eq2db/raids.py

@@ -54,8 +54,8 @@
 import time
 from pathlib import Path
 
+from backend.db_catalogue import BaseCatalogue
 from backend.db_helpers import resolve_db_path
-from backend.eq2db._catalogue import BaseCatalogue
 from backend.sql_loader import load_sql
 
 _SQL = load_sql(__file__)