【Redis】Python连接redis增删改查

• 一、python连接redis：
• • 1.1、首先安装redis模块：
  • 1.2、python连接redis数据库：
  • 1.2、查看表名称是否存在：
  • 1.3、redis新增&更改操作：
  • 1.4、获取redis表下所有key值：
  • 1.5、根据表名称和key值获取对应的value：
  • 1.6、根据数据表名称返回keys、values列表：
  • 1.7、redis根据数据表名称及key值删除对应的value值

说明：平时工作当中为了读取方便都会采取redis缓存的方式那么对于redis的增、删、改、查操作就有需求

pip install redis

import redis
import time
from ReadYaml import *


class operate_redis(object):
    def __init__(self,number):
        # redis数据库连接重试功能和连接超时功能的连接
        self._max_retries_count = 2  # 设置最大重试次数
        self._conn_retries_count = 0  # 初始重试次数
        self._conn_timeout = 10  # 连接超时时间为10秒
        self.db = number  # 连接对应redis数据库db13
        # redis数据库连接
        self._list = []
        self.host = ec_mysql_redis1()['host']
        self.pw = ec_mysql_redis1()['password']
        self.port = ec_mysql_redis1()['port']
        while self._conn_retries_count <= self._max_retries_count:
            try:
                self.redis = redis.Redis(
                    host=self.host,
                    password=self.pw,
                    db=self.db,
                    port=self.port,
                    socket_connect_timeout=self._conn_timeout)
                break
            except:
                print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}redis数据库连接失败")
                self._conn_retries_count += 1
            time.sleep(2)
            continue

    def if_exists(self, name: str):
        # 校验输入的redis数据表名称是否存在
        return self.redis.exists(name)

    def add_redis(self, name: str, key: str, value):
        # 指定redis新增和更改操作，key值没有则新建，有则更新
        try:
            self.redis.hset(name, key, value)
        except Exception as e:
            raise e
        else:
            print("redis写入操作：success")

    def get_keys(self, name: str):
        # 获取redis表下所有key键值
        if operate_redis().if_exists(name=name) == 1:
            try:
                for i in self.redis.hkeys(name):
                    list = self._list
                    list.append(i.decode('utf8'))
            except Exception as e:
                raise e
            else:
                return list
        else:
            print("redis下name有误请检查后重新输入")

    def query_redis(self, name: str, key: str):
        # 输入redis数据表名称及对应的key值精确查找value值
        if operate_redis().if_exists(name=name) == 1:
            try:
                keys = operate_redis().get_keys(name=name)
                if key in keys:
                    value = self.redis.hget(name=name, key=key)
                    return value.decode('utf8')
                    # print(value.decode('utf8'))
                else:
                    print("该name名称下不存在该key值，请查证后重新输入")
            except Exception as e:
                raise e
        else:
            print("redis下name有误请检查后重新输入")

    def get_keys_values(self, name: str):
        # 输入redis数据表的名称返回keys,values列表
        if operate_redis().if_exists(name=name) == 1:
            try:
                keys = operate_redis().get_keys(name=name)
                values = self._list
                for key in keys:
                    value = self.redis.hget(name=name, key=key)
                    values.append(value.decode('utf8'))
                return [keys, values]
            except Exception as e:
                raise e
        else:
            print("redis下name有误请检查后重新输入")

    def deleted_redis(self, name: str, key: str):
        if operate_redis().if_exists(name=name) == 1:
            try:
                keys = operate_redis().get_keys(name=name)
                if key in keys:
                    self.redis.hdel(name, key)
                    print(f"删除{name}下{key}键值对成功")
                else:
                    print("该name名称下不存在该key值，请查证后重新输入")
            except Exception as e:
                raise e
        else:
            print("redis下name有误请检查后重新输入")