即使存在执行此操作的模块(请参阅其他答案),我也尝试编写一个简单的独立版本。这是一个这样的类
KeyValueStore(键和值是字符串):
演示版
from sqlitekeyvaluestore import KeyValueStorekv = KeyValueStore('test.db') # uses SQLiteprint(len(kv)) # 0 itemkv['hello1'] = 'you1'kv['hello2'] = 'you2'kv['hello3'] = 'you3'print(kv['hello1']) # you1print(len(kv)) # 3 itemsdel kv['hello1']print(len(kv)) # 2 items remainingprint('hello1' in kv) # False, it has just been deleted!print('hello3' in kv) # Truekv['hello3'] = 'newvalue' # redefine an already present key/valueprint(kv['hello3']) # newvalueprint(kv.keys()) # ['hello2', 'hello3']print(kv.values()) # ['you2', 'newvalue']print(kv.items()) # [('hello2', 'you2'), ('hello3', 'newvalue')]for k in kv: print(k, kv[k])kv.close() # important to commit代码:sqlitekeyvaluestore.py
import sqlite3class KeyValueStore(dict): def __init__(self, filename=None): self.conn = sqlite3.connect(filename) self.conn.execute("CREATE TABLE IF NOT EXISTS kv (key text unique, value text)") def close(self): self.conn.commit() self.conn.close() def __len__(self): rows = self.conn.execute('SELECT COUNT(*) FROM kv').fetchone()[0] return rows if rows is not None else 0 def iterkeys(self): c = self.conn.cursor() for row in self.conn.execute('SELECt key FROM kv'): yield row[0] def itervalues(self): c = self.conn.cursor() for row in c.execute('SELECt value FROM kv'): yield row[0] def iteritems(self): c = self.conn.cursor() for row in c.execute('SELECt key, value FROM kv'): yield row[0], row[1] def keys(self): return list(self.iterkeys()) def values(self): return list(self.itervalues()) def items(self): return list(self.iteritems()) def __contains__(self, key): return self.conn.execute('SELECt 1 FROM kv WHERe key = ?', (key,)).fetchone() is not None def __getitem__(self, key): item = self.conn.execute('SELECt value FROM kv WHERe key = ?', (key,)).fetchone() if item is None: raise KeyError(key) return item[0] def __setitem__(self, key, value): self.conn.execute('REPLACe INTO kv (key, value) VALUES (?,?)', (key, value)) def __delitem__(self, key): if key not in self: raise KeyError(key) self.conn.execute('DELETE FROM kv WHERe key = ?', (key,)) def __iter__(self): return self.iterkeys()


