Coverage for ramose / cache.py: 100%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import json 

6import sqlite3 

7import time 

8from pathlib import Path 

9 

10 

11class ResultCache: 

12 def __init__(self, directory): 

13 db_dir = Path(directory) 

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

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

16 self._conn.execute( 

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

18 ) 

19 self._conn.commit() 

20 

21 def get(self, key): 

22 row = self._conn.execute( 

23 "SELECT value FROM cache WHERE key = ? AND expires_at > ?", (key, time.time()) 

24 ).fetchone() 

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

26 

27 def set(self, key, value, expire): 

28 self._conn.execute( 

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

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

31 ) 

32 self._conn.commit()