市面上已經有很優秀的嵌入式KV數據庫了，如Berkeley DB。為什么還需要把Sqlite當KV數據庫用呢？原因若干。
1，可能是為了好玩或者純屬無聊
2，可結合關系型數據庫與KV數據庫的優點
3，可利用一些sqlite特性做其他KV數據庫不好做的事情
4，事務管理更方便
5，sqlite更可靠，更流行
實現思路
使用json（或pickle）dump數據，并將數據寫入有KEY（主鍵）和VALUE兩個字段的SQLITE庫表中。參照kv數據庫調用辦法實現外部接口。
主要功能
1，put：寫入key/value數據
2，get：獲取某個key的value
3，put_many：批量寫入key/value數據
4，keys：獲取所有key的列表
5，value：獲取所有value的列表
6，limit：利用SQL語句中limit關鍵字，獲取數據庫中“前”N條KV數據
7，random：利用SQL語句中random關鍵字，從數據庫中隨即獲取N條KV數據
8，has_key：某個key是否存在
9，cursor_execute：執行sql自定義語句
10，其他：items，pop，filter，count等

代碼（KVSqlite.py）：

import os, json, sqlite3 from threading import LockPY3 = os.sys.version_info >= (3,) if PY3:ifilter = filter else:from itertools import ifilterDUMPS = lambda d: json.dumps(d) LOADS = lambda d: json.loads(d)class SDB(object):_DEFAULT_TABLE = '__KVS_DEFAULT_TABLE__'_MEMORY_DB = ":memory:"def __init__(self, filename):if filename is None \or len(filename) < 1 \or filename.lower() == self._MEMORY_DB:self.filename = self._MEMORY_DBelse:self.filename = filenameself._lock = Lock()self._db_init()def _row_factory(self, cursor, row):result = []for idx, col in enumerate(cursor.description):if col[0].lower() in ('k', 'v'):result.append(LOADS(row[idx]))else:result.append(row[idx])return resultdef _db_init(self):_new_table = "CREATE TABLE IF NOT EXISTS {0} ( k PRIMARY KEY,v)".format(self._DEFAULT_TABLE)db = sqlite3.connect(self.filename, timeout=60, check_same_thread=False)db.row_factory = self._row_factorydb.execute(_new_table)self._cursor = db.cursor()self._db = dbdef _statement_init(self):table = self._DEFAULT_TABLEreturn dict(insert="insert or replace into {0}(k,v) values(:1,:2)".format(table),delete="delete from {0} where k=:1".format(table),update="update {0} set v=:1 where k=:2".format(table),clear="delete from {0}".format(table),get="select v from {0} where k=:1".format(table),has_key="select count(1) from {0} where k=:1".format(table),keys="select k from {0}".format(table),values="select v from {0}".format(table),items="select k,v from {0}".format(table),count="select count(*) from {0}".format(table),random="select * from {0} order BY RANDOM() limit :1".format(table),limit="select * from {0} limit :1 offset :2".format(table))_statements = property(_statement_init)del _statement_initdef _insert(self, key, value):try:self._lock.acquire(True)self._cursor.execute(self._statements.get('insert'), (DUMPS(key), DUMPS(value)))finally:self._lock.release()def _update(self, key, value):try:self._lock.acquire(True)self._cursor.execute(self._statements.get('update'), (DUMPS(value), DUMPS(key)))finally:self._lock.release()def _delete(self, key):try:self._lock.acquire(True)self._cursor.execute(self._statements.get('delete'), (DUMPS(key),))finally:self._lock.release()def _clear(self):'''刪除所有數據:return:'''try:self._lock.acquire(True)self._cursor.execute(self._statements.get('clear'))except Exception as e:self._db.rollback()raise efinally:self._lock.release()def keys(self, sort=False, sort_key=None, reverse=False):if sort:return sorted(self.iterkeys(), key=sort_key, reverse=reverse)return list(self.iterkeys())def values(self, sort=False, sort_key=None, reverse=False):if sort:return sorted(self.itervalues(), key=sort_key, reverse=reverse)return list(self.itervalues())def iterkeys(self):try:self._lock.acquire(True)for k in self._cursor.execute(self._statements.get('keys')):yield k[0]finally:self._lock.release()def itervalues(self):try:self._lock.acquire(True)for v in self._cursor.execute(self._statements.get('values')):yield v[0]finally:self._lock.release()def items(self, sort=False, key=None, reverse=False):if sort:return sorted(self.iteritems(), key=key, reverse=reverse)return list(self.iteritems())def iteritems(self):try:self._lock.acquire(True)for k, v in self._cursor.execute(self._statements.get('items')):yield k, vfinally:self._lock.release()def count(self):try:self._lock.acquire(True)return self._cursor.execute(self._statements.get('count')).fetchone()[0]finally:self._lock.release()def has_key(self, key):try:self._lock.acquire(True)r = self._cursor.execute(self._statements.get('has_key'), (DUMPS(key),)).fetchone()[0]return r > 0finally:self._lock.release()def get(self, key):try:self._lock.acquire(True)_key = DUMPS(key)data = self._cursor.execute(self._statements.get('get'), (_key,)).fetchone()if data:return data[0]finally:self._lock.release()def put(self, key, value):try:self._insert(key, value)self._db.commit()except Exception as e:self._db.rollback()raise edef pop(self, key):try:value = self.get(key)self._delete(key)self._db.commit()return valueexcept Exception as e:self._db.rollback()raise edef put_many(self, rows):try:self._lock.acquire(True)if rows and len(rows) > 0:self._cursor.executemany(self._statements.get('insert'),[(DUMPS(k), DUMPS(v)) for k, v in rows])self._db.commit()except Exception as e:self._db.rollback()raise efinally:if self._lock.locked():self._lock.release()def limit(self, limit=1, offset=0):try:self._lock.acquire(True)rows = self._cursor.execute(self._statements.get('limit'), (limit, offset))if limit == 1:return rows.fetchone()return rows.fetchall()finally:self._lock.release()def random(self, limit=1):try:self._lock.acquire(True)rows = self._cursor.execute(self._statements.get('random'), (limit,))if limit == 1:return rows.fetchone()return rows.fetchall()finally:self._lock.release()def filter(self, func):return list(ifilter(func, self.items()))def ifilter(self, func):return ifilter(func, self.iteritems())def cursor_execute(self, sql, parameters=None):'''執行SQL語句，如:SELECT K,V FROM __KVS_DEFAULT_TABLE__ WHERE K LIKE 'ABC%''''try:self._lock.acquire(True)return self._cursor.execute(sql=sql, parameters=parameters)finally:self._lock.release()def close(self):try:self._db.rollback()self._cursor.close()self._db.close()except:pass

調用示例：

from KVSqlite import SDB #打開數據庫 db = SDB('test.sqlite') #寫入單條數據 db.put('first','第一條數據') db.put('second',dict(a=1,b=2,c=[2,3,4])) #獲取數據 db.get('first') #寫入多條數據 db.put_many([[1,2],[3,4],['A','abc']]) #獲取key的列表 db.keys()

總結

以上是生活随笔為你收集整理的把Sqlite当嵌入式KV数据库用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。