wizard/cache.py

   1 """
   2 Provides caching of old results from mass commands, so that
   3 we don't have to do them again.  Successes are cached as
   4 the string "OK"; errors are stored as the string exception name.
   5 Entries are indexed by location.
   6
   7 .. testsetup:: *
   8
   9     from wizard.cache import *
  10 """
  11
  12 import sqlite3
  13
  14 def make(file):
  15     if file:
  16         return SerializedDict(file)
  17     else:
  18         return {}
  19
  20 class SerializedDict(object):
  21     """
  22     Implements a dictionary that serializes itself to disk.  We
  23     use sqlite in order to implement relatively efficient updates.
  24
  25         >>> import os
  26         >>> f = "/tmp/wizard-cache-test.db"
  27         >>> if os.path.exists(f): os.unlink(f)
  28         >>> d = SerializedDict(f)
  29         >>> try:
  30         ...   d["foo"]
  31         ...   raise Exception("didn't raise")
  32         ... except KeyError:
  33         ...   pass
  34         >>> d["foo"] = "baz"
  35         >>> d["foo"]
  36         'baz'
  37         >>> d["foo"] = "bar"
  38         >>> d["foo"]
  39         'bar'
  40         >>> e = SerializedDict(f)
  41         >>> e["foo"]
  42         'bar'
  43         >>> del e["foo"]
  44         >>> "foo" in e
  45         False
  46         >>> os.unlink(f)
  47     """
  48     #: Connection to sqlite database
  49     conn = None
  50     def __init__(self, file):
  51         self.conn = sqlite3.connect(file)
  52         try:
  53             self.conn.execute('create table dict(key string primary key, value string)')
  54         except sqlite3.OperationalError as e:
  55             if e.args[0] != "table dict already exists":
  56                 raise
  57         self.conn.isolation_level = None
  58     def __getitem__(self, key):
  59         row = self.conn.execute('select value from dict where key = ?', (key,)).fetchone()
  60         if row:
  61             return str(row[0])
  62         else:
  63             raise KeyError
  64     def __setitem__(self, key, value):
  65         self.conn.execute('insert or replace into dict (key, value) values (?, ?)', (key, value))
  66     def __delitem__(self, key):
  67         self.conn.execute('delete from dict where key = ?', (key,))
  68     def __contains__(self, key):
  69         return bool(self.conn.execute('select count(*) from dict where key = ?', (key,)).fetchone()[0])