Python脚本如何操作键值数据库

本文目录导读：

1. Redis (最常用的键值数据库)
2. Memcached (内存键值缓存)
3. RocksDB (嵌入式键值存储)
4. LevelDB (另一种嵌入式键值存储)
5. TiKV (分布式键值数据库)
6. 使用连接池（以Redis为例）
7. 异常处理
8. 选择建议

我来详细介绍几种常见的操作键值数据库的Python方法：

Redis (最常用的键值数据库)

安装

pip install redis

基本操作

import redis
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 设置键值
r.set('name', '张三')
r.set('count', 100)
# 获取值
name = r.get('name')  # '张三'
count = r.get('count')  # '100'
# 设置过期时间（秒）
r.setex('temp_key', 3600, '临时数据')
# 批量操作
r.mset({'key1': 'value1', 'key2': 'value2'})
values = r.mget(['key1', 'key2'])  # ['value1', 'value2']
# 检查键是否存在
exists = r.exists('name')  # True/False
# 删除键
r.delete('temp_key')
# 获取所有键
keys = r.keys('*')

高级数据类型

# 哈希 (Hash)
r.hset('user:1', 'name', '李四')
r.hset('user:1', 'age', 25)
user = r.hgetall('user:1')  # {'name': '李四', 'age': '25'}
# 列表 (List)
r.lpush('queue', 'task1', 'task2')
task = r.rpop('queue')
# 集合 (Set)
r.sadd('tags', 'python', 'redis', 'database')
tags = r.smembers('tags')
# 有序集合 (Sorted Set)
r.zadd('scores', {'张三': 95, '李四': 87})
top = r.zrevrange('scores', 0, 2, withscores=True)

Memcached (内存键值缓存)

安装

pip install python-memcached

基本操作

import memcache
# 连接
mc = memcache.Client(['127.0.0.1:11211'])
# 设置值
mc.set('username', 'admin')
mc.set('count', 42, time=3600)  # 设置过期时间
# 获取值
username = mc.get('username')
# 批量操作
mc.set_multi({'key1': 'val1', 'key2': 'val2'})
values = mc.get_multi(['key1', 'key2'])
# 删除
mc.delete('username')
# 增加/减少
mc.incr('count', 5)  # 增加5
mc.decr('count', 3)  # 减少3

RocksDB (嵌入式键值存储)

安装

pip install python-rocksdb

基本操作

import rocksdb
# 打开数据库
db = rocksdb.DB("mydb.rocksdb", rocksdb.Options(create_if_missing=True))
# 写入
db.put(b'key1', b'value1')
db.put(b'key2', b'value2'.encode())
# 读取
value = db.get(b'key1')  # b'value1'
# 批量写入
batch = rocksdb.WriteBatch()
batch.put(b'key3', b'value3')
batch.put(b'key4', b'value4')
batch.delete(b'key1')
db.write(batch)
# 迭代器
it = db.iteritems()
it.seek_to_first()
for key, value in it:
    print(f'{key}: {value}')
# 关闭
del db

LevelDB (另一种嵌入式键值存储)

安装

pip install plyvel

基本操作

import plyvel
# 打开/创建数据库
db = plyvel.DB('mydb.ldb', create_if_missing=True)
# 写入
db.put(b'key1', b'value1')
db.put(b'username', b'admin')
# 读取
value = db.get(b'key1')
# 删除
db.delete(b'key1')
# 批量操作
wb = db.write_batch()
wb.put(b'k1', b'v1')
wb.put(b'k2', b'v2')
wb.delete(b'old_key')
wb.write()
# 遍历
for key, value in db:
    print(f'Key: {key}, Value: {value}')
# 快照（读取一致性视图）
snapshot = db.snapshot()
value = snapshot.get(b'key1')
# 关闭
del db

TiKV (分布式键值数据库)

安装

pip install python-tikv-client

基本操作

from tikv_client import RawClient
# 连接
client = RawClient.connect("127.0.0.1:2379")
# 设置
client.put(b'key1', b'value1')
# 批量设置
client.batch_put([
    (b'key1', b'value1'),
    (b'key2', b'value2')
])
# 获取
value = client.get(b'key1')
# 删除
client.delete(b'key1')
# 扫描
pairs = client.scan(b'key', b'keyz', limit=10)

使用连接池（以Redis为例）

import redis
from redis import ConnectionPool
# 创建连接池
pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=10
)
# 从连接池获取连接
r = redis.Redis(connection_pool=pool)
# 使用上下文管理器
def process_data():
    with redis.Redis(connection_pool=pool) as conn:
        conn.set('key', 'value')
        return conn.get('key')

异常处理

import redis
from redis.exceptions import RedisError
try:
    r = redis.Redis(host='localhost', port=6379)
    # 测试连接
    r.ping()
    # 进行操作
    r.set('key', 'value')
    result = r.get('key')
except ConnectionError as e:
    print(f"连接失败: {e}")
except TimeoutError as e:
    print(f"连接超时: {e}")
except RedisError as e:
    print(f"Redis错误: {e}")
finally:
    r.close()

选择建议

• Redis: 功能最丰富，支持复杂数据结构，适合缓存、消息队列等
• Memcached: 简单快速，纯内存，适合简单缓存场景
• RocksDB/LevelDB: 嵌入式存储，适合本地大量数据存储
• TiKV: 分布式场景，需要高可用性和横向扩展

根据你的具体需求（性能要求、数据持久性、分布式需求等）选择合适的键值数据库。