import sqlite3 import json import logging from typing import Optional import xdg import os import datetime CACHE_DB: Optional[sqlite3.Connection] = None CACHE_PATH = os.path.join(xdg.xdg_cache_home(), 'codigoparallevar', 'ops.sqlite3') def init_db(): global CACHE_DB os.makedirs(os.path.dirname(CACHE_PATH), exist_ok=True) CACHE_DB = sqlite3.connect(CACHE_PATH) cur = CACHE_DB.cursor() cur.execute('''CREATE TABLE IF NOT EXISTS ops( in_val TEXT PRIMARY KEY, code TEXT, out_val TEXT, added_at DateTime ); ''') CACHE_DB.commit() cur.close() def query_cache(in_val, code): if CACHE_DB is None: init_db() assert CACHE_DB is not None cur = CACHE_DB.cursor() cur.execute('''SELECT out_val FROM ops WHERE in_val = ? AND code = ?''', (in_val, code)) # Should return only one result, right? 🤷 results = cur.fetchall() assert len(results) < 2 if len(results) == 0: return None else: return results[0][0] def save_cache(in_val, code, out_val): if CACHE_DB is None: init_db() assert CACHE_DB is not None cur = CACHE_DB.cursor() cur.execute(''' INSERT INTO ops(in_val, code, out_val, added_at) VALUES (?, ?, ?, ?);''', (in_val, code, out_val, datetime.datetime.now())) CACHE_DB.commit() cur.close() def cache(fun): fun_code = fun.__code__.co_code.decode('latin-1') def wrapped(*kargs, **kwargs): in_val = json.dumps({ 'kargs': kargs, 'kwargs': kwargs, 'fun_code': fun_code, }) cache_result = query_cache(in_val, fun_code) found_in_cache = cache_result is not None if not found_in_cache: out_val = fun(*kargs, **kwargs) save_cache(in_val, fun_code, out_val) else: out_val = cache_result logging.info("{} bytes in, {} bytes out (in_cache: {})".format(len(in_val), len(out_val), found_in_cache)) return out_val return wrapped