下載redis命令
pip install redis
連接redis
import redis
# host是redis主機,需要redis服務端和客戶端都起著 redis默認端口是6379
pool = redis.ConnectionPool(host='localhost', port=6379,decode_responses=True)
r = redis.Redis(connection_pool=pool)
操作字符串
# key是"aa" value是"11" 將鍵值對存入redis緩存,有效期300秒
r.set('aa', '11',ex=300)
# key是"bb" value是"22" 將鍵值對存入redis緩存,有效期30000毫秒
r.set('bb', '22',px=30000)
# 取出鍵aa對應的值
print(r.get('aa'))
#批量獲取,取出鍵gender、name對應的值
print(r.mget('aa', 'bb'))
操作hash
# 設置一個哈希“user1”,其中包括name=shamo,age=18,job=tester
r.hset('user1','name','lisa')
r.hset('user1','age',18)
r.hset('user1','job','tester')
# 獲取user1對應的所有值
print(r.hgetall('user1'))
# 批量設置哈希"user2",其中包括name=xiao,age=18,job=tester
r.hmset('user2',{'name':'xiao','age':18,'job':'tester'})
# 批量獲取哈希"user2"中的key為name和age的值
print(r.hmget('user2', 'name', 'age'))
操作列表
# 向列表"list1"中追加數據data1,data2
r.lpush('list1','data1','data2')
# 獲取列表"list1"中所有數據
print(r.lrange('list1',0,-1))
操作集合
# 向列表"set1"中追加數據data1,data2
r.sadd('set1','data1','data2')
# 獲取集合“set1”的所有元素
print(r.smembers('set1'))
操作有序集合
# 向有序集合"zset1"中追加數據data1,data2,其中數字是各自數據的分數,用來排序
r.zadd('zset1',{'data1':14,'data2':13})
# 獲取"zset1"中所有數據
print(r.zrange('zset1',0,-1))
其他操作
r.delete('key1') #刪除某個key
r.exists('key1') #檢查key是否存在
r.expire('user1',10) #設置這個key的有效期為10秒
r.pexpire('user2',10000) #設置這個key的有效期為10000毫秒
r.type('zset1') #獲取某個key對應的數據類型
備注:項目后臺是java語言開發,所以在redis緩存中存儲的基本都是java對象的序列化數據,那么python在獲取java對象的序列化數據后需要進行反序列化轉成對象才能正常獲取數據信息,因此安裝一個第三方庫,用來說java對象的轉換
安裝第三方庫命令
pip install javaobj-py3
然后封裝一個redis方法:
class RedisUtil:
def __init__(self,host,pwd,port=6379,decode_responses=False):
"""
:param host: redis服務器地址
:param pwd: redis密碼
:param port: redis端口 默認6379
:param decode_responses: 默認為False,表示返回的是bytes數據
"""
self.pool = redis.ConnectionPool(host=host,
port=port,password=pwd,encoding_errors='ignore',decode_responses=decode_resp
onses)
self.r = redis.Redis(connection_pool=self.pool)
def get(self,key):
type = self.r.type(key).decode('utf8')
if type=='string':
return self.r.get(key)
elif type=='hash':
return self.r.hgetall(key)
elif type=='zset':
return self.r.zrange(key,0,-1)
elif type=='set':
return self.r.smembers(key)
else:
raise Exception(f'不支持的數據類型{type}或者{key}不存在')```然后在有需要用到redis中的用例中調用:rd = RedisUtil(host='121.42.15.146',pwd='testfan')
res = rd.get('xxxx')
#將從redis中得到的序列化java對象轉換成python對象
res1 = javaobj.loads(res) #res1是個列表如果接口運用到redis比較多,思路:在公共方法里面加一個鉤子函數。例如:
@pytest.fixture(scope=‘session’)
def redis_conn():
redis_util=RedisUtil(host=‘127.0.0.1’, pwd=‘123456’)
return redis_conn
然后在需要用的測試用例里面直接調用res = redis_con.get()