Coverage for ramose / cache.py: 90%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import json 

8import sqlite3 

9import time 

10from pathlib import Path 

11 

12 

13class ResultCache: 

14 def __init__(self, directory: str) -> None: 

15 db_dir = Path(directory) 

16 db_dir.mkdir(parents=True, exist_ok=True) 

17 self._conn = sqlite3.connect(str(db_dir / "cache.db"), check_same_thread=False) 

18 self._conn.execute( 

19 "CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, value TEXT NOT NULL, expires_at REAL NOT NULL)", 

20 ) 

21 self._conn.commit() 

22 

23 def get(self, key: str) -> object: 

24 row = self._conn.execute( 

25 "SELECT value FROM cache WHERE key = ? AND expires_at > ?", 

26 (key, time.time()), 

27 ).fetchone() 

28 return json.loads(row[0]) if row else None 

29 

30 def set(self, key: str, value: object, expire: int) -> None: 

31 self._conn.execute( 

32 "INSERT OR REPLACE INTO cache (key, value, expires_at) VALUES (?, ?, ?)", 

33 (key, json.dumps(value), time.time() + expire), 

34 ) 

35 self._conn.commit() 

36 

37 def clear(self) -> None: 

38 self._conn.execute("DELETE FROM cache") 

39 self._conn.commit()