一个拥有20多年历史,用C语言编写的嵌入式SQL数据库, 似乎不像是并发应用程序键值存储的最佳解决方案。 而且它不是开箱即用的,但是如果配置正确,它的性能会非常出色。 除此之外,它经过了良好的测试、验证和稳定,这是一个数据库存储有价值数据的好特性。
创建 Key-Value 存储表
import sqlite3
# 创建或连接数据库
conn = sqlite3.connect('xx_key_value_store.db')
cursor = conn.cursor()
# 创建 key-value 表
cursor.execute('''
CREATE TABLE IF NOT EXISTS kv_store (
key TEXT PRIMARY KEY,
value TEXT
)
''')
conn.commit()
插入/更新数据
def set_value(key, value):
cursor.execute('''
INSERT OR REPLACE INTO kv_store (key, value)
VALUES (?, ?)
''', (key, value))
conn.commit()
# 示例用法
set_value('name', 'John Doe')
set_value('email', 'john@example.com')
set_value('age', '30')
查询数据
def get_value(key):
cursor.execute('SELECT value FROM kv_store WHERE key = ?', (key,))
result = cursor.fetchone()
return result[0] if result else None
# 示例用法
print(get_value('name')) # 输出: John Doe
print(get_value('phone')) # 输出: None
John Doe None
删除数据
def delete_key(key):
cursor.execute('DELETE FROM kv_store WHERE key = ?', (key,))
conn.commit()
# 示例用法
delete_key('age')
获取所有键值对
def get_all_items():
cursor.execute('SELECT key, value FROM kv_store')
return dict(cursor.fetchall())
# 示例用法
print(get_all_items())
# 输出: {'name': 'John Doe', 'email': 'john@example.com'}
{'name': 'John Doe', 'email': 'john@example.com'}
完整封装类
class KeyValueStore:
def __init__(self, db_file=':memory:'):
self.conn = sqlite3.connect(db_file)
self.cursor = self.conn.cursor()
self._create_table()
def _create_table(self):
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS kv_store (
key TEXT PRIMARY KEY,
value TEXT
)
''')
self.conn.commit()
def __setitem__(self, key, value):
self.cursor.execute('''
INSERT OR REPLACE INTO kv_store (key, value)
VALUES (?, ?)
''', (key, str(value)))
self.conn.commit()
def __getitem__(self, key):
self.cursor.execute('SELECT value FROM kv_store WHERE key = ?', (key,))
result = self.cursor.fetchone()
if result:
return result[0]
raise KeyError(key)
def __delitem__(self, key):
self.cursor.execute('DELETE FROM kv_store WHERE key = ?', (key,))
if self.cursor.rowcount == 0:
raise KeyError(key)
self.conn.commit()
def __contains__(self, key):
self.cursor.execute('SELECT 1 FROM kv_store WHERE key = ?', (key,))
return self.cursor.fetchone() is not None
def close(self):
self.conn.close()
# 上下文管理器支持
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
# 使用示例
with KeyValueStore('my_store.db') as kv:
kv['language'] = 'Python'
kv['version'] = '3.9'
print(kv['language']) # 输出: Python
del kv['version']
Python
高级功能:JSON 值存储
import json
class JSONKeyValueStore(KeyValueStore):
def __setitem__(self, key, value):
super().__setitem__(key, json.dumps(value))
def __getitem__(self, key):
return json.loads(super().__getitem__(key))
# 使用示例
with JSONKeyValueStore('json_store.db') as kv:
kv['config'] = {'theme': 'dark', 'font_size': 14}
print(kv['config']['theme']) # 输出: dark
dark