简介

像操作Python list一样操作Redis list。

数据结构：Redis列表LIST
功能：增删改查

TODO

1. 添加索引操作

代码

redis_dal.py

# -*- coding: utf-8 -*-
# @Author  : XerCis
# @Time    : 2020/3/11 20:00
# @Function: Redis数据访问层

import json
import redis


class RedisDAL():
    def __init__(self, host='localhost', port=6379, db=0, password=None, charset='UTF-8', expire=86400, max=10):
        self.host = host
        self.port = port
        self.db = db
        self.password = password
        self.charset = charset
        self.db = redis.StrictRedis(host, port, db, password, charset=charset, decode_responses=True)
        self.expire = expire  # 过期时间，默认86400(24小时)
        self.max = max  # 每个键允许存储的最大条数，默认10

    def __repr__(self):
        return self.__str__()

    def __str__(self):
        return '{} {}'.format(__class__.__name__, self.__dict__)

    def __len__(self):
        '''库中键的数量'''
        return self.db.dbsize()

    @property
    def expire(self):
        return self._expire

    @expire.setter
    def expire(self, expire):
        if not isinstance(expire, int):
            raise TypeError
        self._expire = expire

    @property
    def max(self):
        return self._max

    @max.setter
    def max(self, max):
        if not isinstance(max, int):
            raise TypeError
        self._max = max

    def create(self, name, value):
        '''增，含新增和添加

        :param name: 键名
        :type name: str
        :param value: 可被json转换的任意值
        :return 是否成功
        :rtype: bool
        '''
        value_str = json.dumps(value, ensure_ascii=False)
        try:
            if self.db.llen(name) == 0:
                # 空的话右插并设置过期时间
                self.db.rpush(name, value_str)
            else:
                # 非空
                if self.db.llen(name) < self.max:
                    # 还能放
                    self.db.rpush(name, value_str)
                else:
                    # 放不下左弹再右插
                    self.db.lpop(name)
                    self.db.rpush(name, value_str)
            self.db.expire(name, self.expire)
        except Exception as e:
            print(str(e))
            return False
        return True

    def read(self, name, start=0, end=-1):
        '''查

        :param name: 键名
        :param start: 起点
        :param end: 终点
        :return: 指定范围的列表
        :rtype: list
        '''
        result = self.db.lrange(name, start, end)
        result = [json.loads(i) for i in result]
        return result

    def update(self, name, value, index=-1):
        '''改

        :param name: 键名
        :param value: 可被json转换的任意值
        :param index: 下标，默认修改最后一个
        :return 是否成功
        :rtype: bool
        '''
        value_str = json.dumps(value, ensure_ascii=False)
        try:
            result = self.db.lset(name, index, value_str)
        except Exception as e:
            print(str(e))
            return False
        return result

    def delete(self, name, num=1):
        '''删，优先删除旧数据

        :param name: 键名
        :param num: 条数
        :return: 删掉的内容
        :rtype: list or False
        '''
        if name == '*':
            # 防止SQL注入
            return False
        result = []
        if self.db.exists(name) == False:
            # 不存在该键
            return False
        exist_len = self.db.llen(name)  # 剩余条数
        num = min(num, exist_len, self.max)  # 能取的最大条数
        try:
            for i in range(num):
                value_str = self.db.lpop(name)  # 左弹
                value = json.loads(value_str)
                result.append(value)
        except Exception as e:
            print(str(e))
            return False
        return result

    def length(self, name):
        '''给定键查询长度

        :param name: 键名
        :return: 长度
        :rtype: int
        '''
        return self.db.llen(name)

    def get_names(self):
        '''所有键名'''
        return self.db.scan()[1]

    def empty(self):
        '''清空'''
        command = input(r'数据库共有{}条数据，确认清空请输入（确认）: '.format(len(self)))
        if command == '确认':
            if self.db.flushdb():
                print('清空成功')
        else:
            print('清空失败')


if __name__ == '__main__':
    '''初始化'''
    redisDAL = RedisDAL()
    print(redisDAL)

    '''设置过期时间和最大条数'''
    redisDAL.expire = 10000  # 过期时间，单位s
    redisDAL.max = 20  # 每个键允许存储的最大条数
    print('过期时间: {}s'.format(redisDAL.expire))
    print('每个键存储的最大条数: {}'.format(redisDAL.max))

    '''增'''
    name = '小明'
    value = {'语文': 70, '数学': 80, '英语': 90}
    redisDAL.create(name, value)  # 新增
    value = {'物理': 70, '化学': 80, '生物': 90}
    redisDAL.create(name, value)  # 添加

    name = '小红'
    value = {'语文': 80, '数学': 90, '英语': 100}
    redisDAL.create(name, value)  # 新增

    names = redisDAL.get_names()
    num = len(redisDAL)  # 数据库的键的数量
    print('数据库共有{}个键，分别是: {}'.format(num, names))

    '''查'''
    name = '小明'
    num = redisDAL.length(name)  # 键的对应值的数量
    print('小明的成绩共{}条，分别是: {}'.format(num, redisDAL.read(name)))

    '''改'''
    name = '小明'
    value = {'语文': 100, '数学': 100, '英语': 100}
    redisDAL.update(name, value, index=0)
    print('小明的成绩修改后: {}'.format(redisDAL.read(name)))

    '''删'''
    name = '小明'
    num = 1
    result = redisDAL.delete(name, num)
    print('小明的成绩删除{}条: {}'.format(num, result))
    print('小明的成绩删除后: {}'.format(redisDAL.read(name)))

    '''清空'''
    redisDAL.empty()

参考文献

1. Redis入门
2. Python 对 JSON 中的特殊类型进行 Encoder

原文链接:https://blog.csdn.net/lly1122334/article/details/104815145