Returns True if NEW, False if already present (so callers can count deltas).
(self, key: str, data: dict)
| 86 | self.db.execute("CREATE TABLE IF NOT EXISTS items " |
| 87 | "(key TEXT PRIMARY KEY, data TEXT, seen_at REAL)") |
| 88 | def upsert(self, key: str, data: dict) -> bool: |
| 89 | """Returns True if NEW, False if already present (so callers can count deltas).""" |
| 90 | cur = self.db.execute("SELECT 1 FROM items WHERE key=?", (key,)) |
| 91 | is_new = cur.fetchone() is None |
| 92 | self.db.execute("INSERT INTO items(key,data,seen_at) VALUES(?,?,?) " |
| 93 | "ON CONFLICT(key) DO UPDATE SET data=excluded.data, seen_at=excluded.seen_at", |
| 94 | (key, json.dumps(data, ensure_ascii=False), time.time())) |
| 95 | self.db.commit() |
| 96 | return is_new |
| 97 | def get(self, key: str): |
| 98 | row = self.db.execute("SELECT data FROM items WHERE key=?", (key,)).fetchone() |
| 99 | return json.loads(row[0]) if row else None |
no test coverage detected