Python调用Redis

深碍√TFBOYSˉ_ 2023-08-17 17:43 104阅读 0赞

1 #!/usr/bin/env python
      2 # -*- coding:utf-8 -*-
      3 # *************************************
      4 # @Time    : 2019/8/12
      5 # @Author  : Zhang Fan
      6 # @Desc    : Library
      7 # @File    : MyRedis.py
      8 # @Update  : 2019/8/23
      9 # *************************************
     10 import redis
     11 
     12 
     13 class MyRedis(object):
     14     """
     15     ===================================================================
     16     =====================       MyRedis        ========================
     17     ===================================================================
     18     """
     19     def __init__(self):
     20         self.redis_conn = None
     21         self.redis_db = None
     22 
     23     def connect_to_redis(self, redis_host, redis_port=6379, db=0, password=None):
     24         """
     25         连接到Redis服务器
     26         """
     27         self.redis_db = db
     28         print('Executing : Connect To Redis | host={0}, port={1}, db={2}, password={3}'
     29                      .format(redis_host, redis_port, self.redis_db, password))
     30         try:
     31             self.redis_conn = redis.StrictRedis(
     32                 host=redis_host, port=redis_port, db=self.redis_db, password=password)
     33         except Exception as ex:
     34             logger.error(str(ex))
     35             raise Exception(str(ex))
     36 
     37     def redis_key_should_be_exist(self, name):
     38         """
     39         验证redis存在指定键
     40         """
     41         if not self.redis_conn.exists(name):
     42             logger.error(("Redis of db%s doesn't exist in key [ %s ]." % (self.redis_db, name)))
     43             raise AssertionError
     44 
     45     def redis_key_should_not_be_exist(self, name):
     46         """
     47         验证redis不存在指定键
     48         """
     49         if self.redis_conn.exists(name):
     50             logger.error(("Redis of db%s exist in key [ %s ]." % (self.redis_db, name)))
     51             raise AssertionError
     52 
     53     def getkeys_from_redis_bypattern(self, pattern, field=None):
     54         """
     55         获取redis所有键值
     56         """
     57         keys_list = list()
     58         print('Executing : Getall Key | %s' % pattern)
     59         if field is None:
     60             return self.redis_conn.keys(pattern)
     61         else:
     62             keys = self.redis_conn.keys(pattern)
     63             for key in keys:
     64                 if not self.redis_conn.hget(key, field) is None:
     65                     keys_list.append(key)
     66             return keys_list
     67 
     68     # ==========================  String Type =============================
     69     def get_from_redis(self, name):
     70         """
     71         获取redis数据
     72         """
     73         print('Executing : Get Key | %s' % name)
     74         return self.redis_conn.get(name)
     75 
     76     def del_from_redis(self, name):
     77         """
     78         删除redis中的任意数据类型
     79         """
     80         return self.redis_conn.delete(name)
     81 
     82     def set_to_redis(self, name, data, expire_time=0):
     83         """
     84         设置redis执行key的值
     85         """
     86         return self.redis_conn.set(name, data, expire_time)
     87 
     88     def append_to_redis(self, name, value):
     89         """
     90         添加数据到redis
     91         """
     92         return self.redis_conn.append(name, value)
     93 
     94         # ==========================  Hash Type  ==========================
     95     def hgetall_from_redis(self, name):
     96         """
     97         获取redis hash所有数据
     98         """
     99         print('Executing : Hgetall Key | %s' % name)
    100         return self.redis_conn.hgetall(name)
    101 
    102     def hget_from_redis(self, name, key):
    103         """
    104         获取redis hash指定key数据
    105         """
    106         print('Executing : Hget Key | %s' % name)
    107         return self.redis_conn.hget(name, key)
    108 
    109     def hset_to_redis(self, name, key, data):
    110         """
    111         设置redis指定key的值
    112         """
    113         print(('Executing : Hset Redis | name={0}, key={1}, data={2}'
    114                      .format(name, key, data)))
    115         return self.redis_conn.hset(name, key, data)
    116 
    117     def hdel_to_redis(self, name, *keys):
    118         """
    119         删除redis指定key的值
    120         """
    121         print('Executing : Hdel Key | ', *keys)
    122         self.redis_conn.hdel(name, *keys)
    123 
    124     # =========================  ZSet Type  ================================
    125     def get_from_redis_zscore(self, name, values):
    126         """
    127         获取name对应有序集合中 value 对应的分数
    128         """
    129         try:
    130             return int(self.redis_conn.zscore(name, values))
    131         except:
    132             return self.redis_conn.zscore(name, values)
    133 
    134     def get_from_redis_zrange(self, name, start=0, end=10):
    135         """
    136         按照索引范围获取name对应的有序集合的元素
    137         """
    138         return self.redis_conn.zrange(name, start, end, desc=False, withscores=True, score_cast_func=int)
    139 
    140     def del_from_redis_zrem(self, name, values):
    141         """
    142         删除name对应的有序集合中值是values的成员
    143         """
    144         return self.redis_conn.zrem(name, values)
    145 
    146     def add_from_redis_zadd(self, name, value, score):
    147         """
    148         在name对应的有序集合中添加一条。若值存在，则修改对应分数。
    149         """
    150         return self.redis_conn.zadd(name, {value: score})
    151 
    152     def count_from_redis_zcard(self, name):
    153         """
    154         获取name对应的有序集合元素的数量
    155         """
    156         return self.redis_conn.zcard(name)
    157 
    158 
    159 if __name__ == '__main__':
    160     print('This is test.')
    161     mr = MyRedis()

转载于:https://www.cnblogs.com/leozhanggg/p/11401444.html