使用SQLite作为键:值存储

正如在这里的评论和其他问题中所建议的那样,SQLite可以完全用作持久性密钥:值存储.

你将如何定义一个类(或只是包装函数),使用一个键:使用SQLite的值存储将如下所示:

kv = Keyvaluestore('/test.db')
kv['hello'] = 'hi'        # set
print kv['hello']         # get
print 'blah' in kv        # answer: False because there's no key 'blah' in the store
kv.close()

？

即使存在执行此操作的模块(请参阅其他答案),我也尝试编写一个简单的自包含版本.这是一个类KeyValueStore(键和值是字符串),其工作方式如下:

用法

kv = KeyValueStore('test.db')    # uses SQLite
kv['hello1'] = 'you1'
kv['hello2'] = 'you2'
kv['hello3'] = 'you3'
print kv['hello1']          # you1
print len(kv)               # 3 items

del kv['hello1']
print len(kv)               # 2 items remaining

print 'hello1' in kv        # False, it has just been deleted!
print 'hello3' in kv        # True

kv['hello3'] = 'newvalue'   # redefine an already present key/value
print kv['hello3']

print kv.keys()             # [u'hello2', u'hello3']
print kv.values()           # [u'you2', u'you3']
print kv.items()            # [(u'hello2', u'you2'), (u'hello3', u'you3')]

for k, v in kv:             # looping over the KeyValueStore
    print k, v              # hello3 you3 // hello2 you2

码

import sqlite3
from UserDict import DictMixin as DictClass

class KeyValueStore(DictClass):
    def __init__(self, filename=None):
        self.cOnn= sqlite3.connect(filename)
        self.conn.execute("CREATE TABLE IF NOT EXISTS kv (key text unique, value text)")
        self.c = self.conn.cursor()

    def close():
        self.conn.commit()
        self.conn.close()

    def __len__(self):
        self.c.execute('SELECT COUNT(*) FROM kv')
        rows = self.c.fetchone()[0]
        return rows if rows is not None else 0

    def iterkeys(self):
        c1 = self.conn.cursor()
        for row in c1.execute('SELECT key FROM kv'):
            yield row[0]

    def itervalues(self):
        c2 = self.conn.cursor()
        for row in c2.execute('SELECT value FROM kv'):
            yield row[0]

    def iteritems(self):
        c3 = self.conn.cursor()
        for row in c3.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):
        self.c.execute('SELECT 1 FROM kv WHERE key = ?', (key,))
        return self.c.fetchone() is not None

    def __getitem__(self, key):
        self.c.execute('SELECT value FROM kv WHERE key = ?', (key,))
        item = self.c.fetchone()
        if item is None:
            raise KeyError(key)
        return item[0]

    def __setitem__(self, key, value):
        self.c.execute('REPLACE INTO kv (key, value) VALUES (?,?)', (key, value))
        self.conn.commit()

    def __delitem__(self, key):
        if key not in self:
            raise KeyError(key)
        self.c.execute('DELETE FROM kv WHERE key = ?', (key,))
        self.conn.commit()

    def __iter__(self):
        return self.iteritems()

已经有sqlitedict可以满足您的所有需求.

从文档:

>>> from sqlitedict import SqliteDict
>>> mydict = SqliteDict('./my_db.sqlite', autocommit=True)
>>> mydict['some_key'] = any_picklable_object
>>> print mydict['some_key']  # prints the new value
>>> for key, value in mydict.iteritems():
>>>     print key, value
>>> print len(mydict) # etc... all dict functions work
>>> mydict.close()