""" Provides caching of old results from mass commands, so that we don't have to do them again. Successes are cached as the string "OK"; errors are stored as the string exception name. Entries are indexed by location. .. testsetup:: * from wizard.cache import * """ import sqlite3 def make(file=None): """ Generates a :class:`SerializedDict` if ``file`` is not ``None``, and a normal dictionary otherwise. """ if file: return SerializedDict(file) else: return {} class SerializedDict(object): """ Implements a dictionary that serializes itself to disk. We use sqlite in order to implement relatively efficient updates. >>> import os >>> f = "/tmp/wizard-cache-test.db" >>> if os.path.exists(f): os.unlink(f) >>> d = SerializedDict(f) >>> try: ... d["foo"] ... raise Exception("didn't raise") ... except KeyError: ... pass >>> d["foo"] = "baz" >>> d["foo"] 'baz' >>> d["foo"] = "bar" >>> d["foo"] 'bar' >>> e = SerializedDict(f) >>> e["foo"] 'bar' >>> del e["foo"] >>> "foo" in e False >>> os.unlink(f) """ #: Connection to sqlite database conn = None def __init__(self, file): self.conn = sqlite3.connect(file) try: self.conn.execute('create table dict(key string primary key, value string)') except sqlite3.OperationalError as e: if e.args[0] != "table dict already exists": raise self.conn.isolation_level = None def __getitem__(self, key): row = self.conn.execute('select value from dict where key = ?', (key,)).fetchone() if row: return str(row[0]) else: raise KeyError def __setitem__(self, key, value): self.conn.execute('insert or replace into dict (key, value) values (?, ?)', (key, value)) def __delitem__(self, key): self.conn.execute('delete from dict where key = ?', (key,)) def __contains__(self, key): return bool(self.conn.execute('select count(*) from dict where key = ?', (key,)).fetchone()[0])