new-codigoparallevar/scripts/ops_cache.py

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