封装redis
Posted 测试的自我放逐
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了封装redis相关的知识，希望对你有一定的参考价值。
import redis
# r = redis.Redis()
class MyRedis():
def __init__(self,ip,password,port=6379,db=0):
#构造函数 类在实例化的时候会执行。
try:
self.r = redis.Redis(host=ip,password=password,port=port,db=db)
# r = redis.ConnectionPool(host=ip,password=password,port=port,db=db)
except Exception as e:
print(‘redis连接失败，错误信息%s‘%e)
def str_get(self,k):
res = self.r.get(k)
if res:
return res.decode() #bytes类型转化为字符串。
def str_set(self,k,v,time=None):
self.r.set(k,v,time) # 不出异常不需要
def delete(self,k):
tag = self.r.exists(k) # r.exists 用来判断这个key是否存在
if tag:
self.r.delete(k)
print(‘删除成功‘)
else:
print(‘这个key不存在‘)
def hash_get(self,name,k):
res = self.r.hget(name,k)
if res:
return res.decode()
def hash_set(self,name,k,v):
self.r.hset(name,k,v)
def hash_getall(self,name):
data = {}
# {b‘12‘: b‘1212‘, b‘3‘: b‘sdad‘, b‘4‘: b‘asdadsa‘}
res = self.r.hgetall(name) #获取的结果是一个字典
if res:
for k,v in res.items():
k = k.decode() #对于返回的结果转化为字符串
v = v.decode()
data[k]=v #加入字典
return data
def hash_del(self,name,k): #尝试删除不存在 不报错 ，所以不需要try
res = self.r.hdel(name,k)
if res:
print(‘删除成功‘)
return 1 #删除成功会返回1 返回给该函数。
else:
print(‘删除失败，该key不存在‘)
return 0
@property #属性方法 可以a.clean_redis直接调用
def clean_redis(self):
self.r.flushdb() #清空redis 使用 flushdb()用来清空库
print(‘清空redis成功！‘)
return 0
####################################
my = MyRedis(‘118.24.xxx‘,‘HK139232‘) #不会报错 try方法没效果，其实有一个 redis.ConnectionPool 的方法可以？？？找一下
my.str_get(‘sss‘)
以上是关于封装redis的主要内容，如果未能解决你的问题，请参考以下文章
Redis操作List工具类封装，Java Redis List命令封装