168 lines
5.6 KiB
Python
168 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Redis 数据库操作模块
|
||
支持基本的键值操作
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import json
|
||
|
||
# 避免文件名冲突，确保导入正确的redis库
|
||
try:
|
||
from redis import Redis as RedisClient
|
||
from redis import ConnectionError as RedisConnectionError
|
||
except ImportError as e:
|
||
print(f"错误: 无法导入 redis 库，请确保已安装: pip install redis")
|
||
print(f"详细错误: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
class RedisDatabase:
|
||
"""Redis 数据库连接和操作类"""
|
||
|
||
def __init__(self, host='localhost', port=6379, db=0, password=None):
|
||
"""
|
||
初始化 Redis 连接参数
|
||
:param host: Redis 主机地址
|
||
:param port: Redis 端口
|
||
:param db: 数据库编号（0-15）
|
||
:param password: 密码（可选）
|
||
"""
|
||
self.host = host
|
||
self.port = port
|
||
self.db = db
|
||
self.password = password
|
||
self.client = None
|
||
|
||
def connect(self):
|
||
"""建立数据库连接"""
|
||
try:
|
||
self.client = RedisClient(
|
||
host=self.host,
|
||
port=self.port,
|
||
db=self.db,
|
||
password=self.password,
|
||
decode_responses=True,
|
||
socket_connect_timeout=5
|
||
)
|
||
# 测试连接
|
||
self.client.ping()
|
||
return True, "Redis 连接成功"
|
||
except RedisConnectionError as e:
|
||
return False, f"Redis 连接失败: 无法连接到 Redis 服务器 - {str(e)}"
|
||
except Exception as e:
|
||
return False, f"Redis 连接失败: {str(e)}"
|
||
|
||
def execute(self, command):
|
||
"""
|
||
执行 Redis 命令
|
||
:param command: Redis 命令字符串
|
||
示例: "SET key value" 或 "GET key"
|
||
"""
|
||
if not self.client:
|
||
return False, "未连接到数据库"
|
||
|
||
try:
|
||
# 解析命令
|
||
parts = command.strip().split()
|
||
if not parts:
|
||
return False, "命令不能为空"
|
||
|
||
cmd = parts[0].upper()
|
||
args = parts[1:]
|
||
|
||
# 执行常用命令
|
||
if cmd == 'SET' and len(args) >= 2:
|
||
key = args[0]
|
||
value = ' '.join(args[1:])
|
||
result = self.client.set(key, value)
|
||
return True, "OK" if result else "FAIL"
|
||
|
||
elif cmd == 'GET' and len(args) >= 1:
|
||
key = args[0]
|
||
result = self.client.get(key)
|
||
return True, result if result is not None else "(nil)"
|
||
|
||
elif cmd == 'DEL' and len(args) >= 1:
|
||
result = self.client.delete(*args)
|
||
return True, f"删除了 {result} 个键"
|
||
|
||
elif cmd == 'EXISTS' and len(args) >= 1:
|
||
result = self.client.exists(*args)
|
||
return True, f"存在 {result} 个键"
|
||
|
||
elif cmd == 'KEYS' and len(args) >= 1:
|
||
pattern = args[0]
|
||
result = self.client.keys(pattern)
|
||
return True, result
|
||
|
||
elif cmd == 'HSET' and len(args) >= 3:
|
||
key = args[0]
|
||
field = args[1]
|
||
value = ' '.join(args[2:])
|
||
result = self.client.hset(key, field, value)
|
||
return True, f"设置字段成功: {result}"
|
||
|
||
elif cmd == 'HGET' and len(args) >= 2:
|
||
key = args[0]
|
||
field = args[1]
|
||
result = self.client.hget(key, field)
|
||
return True, result if result is not None else "(nil)"
|
||
|
||
elif cmd == 'HGETALL' and len(args) >= 1:
|
||
key = args[0]
|
||
result = self.client.hgetall(key)
|
||
return True, result
|
||
|
||
elif cmd == 'LPUSH' and len(args) >= 2:
|
||
key = args[0]
|
||
result = self.client.lpush(key, *args[1:])
|
||
return True, f"列表长度: {result}"
|
||
|
||
elif cmd == 'LRANGE' and len(args) >= 3:
|
||
key = args[0]
|
||
start = int(args[1])
|
||
stop = int(args[2])
|
||
result = self.client.lrange(key, start, stop)
|
||
return True, result
|
||
|
||
elif cmd == 'SADD' and len(args) >= 2:
|
||
key = args[0]
|
||
result = self.client.sadd(key, *args[1:])
|
||
return True, f"添加了 {result} 个元素"
|
||
|
||
elif cmd == 'SMEMBERS' and len(args) >= 1:
|
||
key = args[0]
|
||
result = self.client.smembers(key)
|
||
return True, list(result)
|
||
|
||
elif cmd == 'PING':
|
||
result = self.client.ping()
|
||
return True, "PONG" if result else "FAIL"
|
||
|
||
elif cmd == 'FLUSHDB':
|
||
result = self.client.flushdb()
|
||
return True, "OK" if result else "FAIL"
|
||
|
||
elif cmd == 'DBSIZE':
|
||
result = self.client.dbsize()
|
||
return True, f"数据库键数量: {result}"
|
||
|
||
else:
|
||
# 尝试执行原始命令
|
||
result = self.client.execute_command(*parts)
|
||
return True, result
|
||
|
||
except Exception as e:
|
||
return False, f"执行失败: {str(e)}"
|
||
|
||
def close(self):
|
||
"""关闭数据库连接"""
|
||
if self.client:
|
||
self.client.close()
|
||
self.client = None
|
||
return "Redis 连接已关闭"
|
||
return "连接已经关闭"
|