Jonathan
034ae40084
The worker's MB browser (discography + artist search, used by monitor sweeps, library scans, and Last.fm discovery) re-fetched every time. Route it through the same ApiCache table + logical keys the web app uses, so a discography one process fetched warms the other, and repeat lookups stop hitting MB. - apicache.py cached_api(): read-through, serve-stale-on-outage, never caches null, best-effort (a cache-DB error degrades to a live fetch, never breaks scan/sweep/discovery). Its own DEDICATED autocommit connection so it can't commit the worker's in-flight transaction, and autocommit avoids freezing Postgres now() (which would break TTL math). - CachingMbBrowser decorates MusicBrainzBrowser, serializing to the SAME camelCase JSON shape the web writes (round-trip tested). registry.build_browser wraps it when a DSN is present; the Last.fm source's browser rides it too. - Hourly prune drops ApiCache rows > 90d so the table stays bounded. - worker 213 tests / 7-skip. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Read-through cache over the shared ApiCache table (same table + logical keys the web app
|
|
uses, so web and worker share entries). TTL applied here from fetchedAt; serves a stale row
|
|
if the fresh fetch fails; never caches None.
|
|
|
|
Best-effort: all DB access is guarded so a cache-layer problem NEVER breaks the caller
|
|
(scan/sweep/discovery) — on a DB error it just degrades to a live fetch. It is given a
|
|
DEDICATED connection (not the worker's main conn) so its commits can't commit the worker's
|
|
in-flight transaction."""
|
|
from typing import Callable, TypeVar
|
|
|
|
import psycopg
|
|
from psycopg.types.json import Jsonb
|
|
|
|
T = TypeVar("T")
|
|
|
|
DAY = 86_400
|
|
WEEK = 7 * 86_400
|
|
MONTH = 30 * 86_400
|
|
|
|
|
|
def _read(conn, key: str):
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT json, extract(epoch from (now() - "fetchedAt")) FROM "ApiCache" WHERE key = %s',
|
|
(key,),
|
|
)
|
|
return cur.fetchone()
|
|
except Exception:
|
|
try:
|
|
conn.rollback()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _write(conn, key: str, value) -> None:
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'INSERT INTO "ApiCache" (key, json, "fetchedAt") VALUES (%s, %s, now()) '
|
|
'ON CONFLICT (key) DO UPDATE SET json = EXCLUDED.json, "fetchedAt" = now()',
|
|
(key, Jsonb(value)),
|
|
)
|
|
conn.commit()
|
|
except Exception:
|
|
try:
|
|
conn.rollback()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def cached_api(conn, key: str, ttl_sec: int, fetch_fresh: Callable[[], T]) -> T:
|
|
row = _read(conn, key)
|
|
if row is not None and row[1] < ttl_sec:
|
|
return row[0]
|
|
try:
|
|
fresh = fetch_fresh()
|
|
except Exception:
|
|
if row is not None:
|
|
return row[0] # serve stale on an upstream outage
|
|
raise
|
|
if fresh is not None:
|
|
_write(conn, key, fresh)
|
|
return fresh
|
|
|
|
|
|
def norm_key(s: str) -> str:
|
|
"""Match the web side's normKey (trim + lowercase) so search keys don't fragment."""
|
|
return s.strip().lower()
|
|
|
|
|
|
def open_cache_conn(dsn: str):
|
|
"""A dedicated AUTOCOMMIT connection for cache I/O, or None if it can't be opened (the
|
|
cache then simply no-ops). Kept separate from the worker's main connection. Autocommit is
|
|
essential: a read that hits the TTL returns without committing, and a non-autocommit conn
|
|
would leave that transaction open — freezing Postgres `now()` at the transaction start
|
|
(now() = transaction timestamp), which would break every subsequent TTL comparison."""
|
|
try:
|
|
return psycopg.connect(dsn, autocommit=True)
|
|
except Exception:
|
|
return None
|