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

wen 实用脚本 2026-07-10 2

本文目录导读：

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

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

我来详细介绍几种常见的操作键值数据库的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: 分布式场景，需要高可用性和横向扩展

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