Python - Python操作Redis

安装Redis可参考

Redis-入门简介-CSDN博客

在Python中接入Redis数据库通常使用redis-py这个库

一、安装Redis

首先，需要安装redis-py库。通过pip来安装

pip install redis

二、连接Redis

'''
Redis连接操作
'''
import redisdef redis_connect():try:redisClient = redis.Redis(host='127.0.0.1', port=6379, password=None)# <redis.client.Redis(<redis.connection.ConnectionPool(<redis.connection.Connection(host=127.0.0.1,port=6379,db=0)>)>)>print(redisClient)# 设置key为‘name’的值以及过期时间redisClient.set('name', 'zhangsan', ex=300)# 获取key为‘name’的过期时间print(redisClient.ttl('name'))  # 300# 获取key为‘name’的值print(redisClient.get('name'))  # b'zhangsan';前面的'b'代表字节redisClient.set('nickname', '张三', ex=500)print(redisClient.get('nickname'))  # b'\xe5\xbc\xa0\xe4\xb8\x89'print(redisClient.get('nickname').decode())  # 张三redisClient.set('num', '10')print(redisClient.incr('num'))  # 11print(redisClient.incrby('num', 5))  # 16print(int(redisClient.get('num').decode()))  # 16# 使用哈希表redisClient.hset('user', 'id', '1001')redisClient.hset('user', 'name', '张三')print(redisClient.hgetall('user'))  # {b'id': b'1001', b'name': b'\xe5\xbc\xa0\xe4\xb8\x89'}print(redisClient.hget('user', 'name').decode())  # 张三# 使用列表redisClient.lpush('list', 'a', 'b', 'c')# 获取列表中的所有元素print(redisClient.lrange('list', 0, -1))  # [b'c', b'b', b'a']# 使用集合redisClient.sadd('myset', 'a')redisClient.sadd('myset', 'b')redisClient.smembers('myset')  # 获取集合中的所有元素except Exception as ex:print(ex.args)finally:redisClient.close()if __name__ == '__main__':redis_connect()

使用连接池（可选）

为了提高性能和效率，特别是在多线程环境下，建议使用连接池

'''
Redis连接池
'''
import redisdef redis_connect():"""redis连接:return:"""try:pool = redis.ConnectionPool(host='127.0.0.1', port=6379, decode_responses=True)redisClient = redis.Redis(connection_pool=pool)redisClient.set('nickname', '张三', ex=500)print(redisClient.get('nickname'))  # b'\xe5\xbc\xa0\xe4\xb8\x89'print(redisClient.get('nickname').decode())  # 张三except Exception as ex:print(ex)finally:redisClient.close()pool.close()if __name__ == '__main__':redis_connect()

在使用连接池的情况下，通常不需要手动关闭每个连接，连接池会自动管理连接。如果你使用的是连接池，只需在不再需要时关闭整个连接池

    finally:redisClient.close()pool.close()

三、封装连接Redis数据库工具

'''
1、初始化连接：在 __init__ 中，我们通过 redis.StrictRedis 初始化 Redis 连接，可以根据需要传递不同的 Redis 服务器信息。
2、基本操作封装：
（1）set 和 get 分别用于设置和获取键值。
（2）delete 用于删除指定的键。
（3）exists 用于检查键是否存在。
3、批量操作：提供了 set_multiple 和 get_multiple 方法来处理批量的 Redis 操作。
4、自增和自减：封装了 increment 和 decrement 方法来处理数字类型的键值的增减。
5、哈希表操作：通过 hset, hget, 和 hgetall 封装了对 Redis 哈希表的操作
'''import redis
from typing import Optional, Union, Anyclass RedisHelper:def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):"""初始化 RedisHelper 类，设置 Redis 连接信息:param host: Redis 服务器主机，默认是 localhost:param port: Redis 服务器端口，默认是 6379:param db: Redis 数据库，默认是 0"""self.redis_client = redis.StrictRedis(host=host, port=port, db=db, decode_responses=True)def set(self, key: str, value: Union[str, int, float], ex: Optional[int] = None) -> bool:"""设置键值对，如果设置了 ex 参数，则键值会在过期时间后自动删除:param key: 键:param value: 值，可以是字符串、整数或浮点数:param ex: 过期时间（秒），默认不设置过期时间:return: 是否成功"""try:if ex:self.redis_client.setex(key, ex, value)else:self.redis_client.set(key, value)return Trueexcept Exception as e:print(f"Error setting key {key}: {e}")return Falsedef get(self, key: str) -> Optional[str]:"""获取指定键的值:param key: 键:return: 键对应的值，如果键不存在，则返回 None"""try:return self.redis_client.get(key)except Exception as e:print(f"Error getting key {key}: {e}")return Nonedef delete(self, key: str) -> bool:"""删除指定键:param key: 键:return: 是否成功"""try:self.redis_client.delete(key)return Trueexcept Exception as e:print(f"Error deleting key {key}: {e}")return Falsedef exists(self, key: str) -> bool:"""检查指定的键是否存在:param key: 键:return: 键是否存在"""try:return self.redis_client.exists(key)except Exception as e:print(f"Error checking existence of key {key}: {e}")return Falsedef set_multiple(self, mapping: dict, ex: Optional[int] = None) -> bool:"""批量设置多个键值对:param mapping: 键值对字典:param ex: 过期时间（秒），可选:return: 是否成功"""try:if ex:for key, value in mapping.items():self.redis_client.setex(key, ex, value)else:self.redis_client.mset(mapping)return Trueexcept Exception as e:print(f"Error setting multiple keys: {e}")return Falsedef get_multiple(self, keys: list) -> dict:"""批量获取多个键的值:param keys: 键列表:return: 键值对字典"""try:values = self.redis_client.mget(keys)return dict(zip(keys, values))except Exception as e:print(f"Error getting multiple keys: {e}")return {}def increment(self, key: str, amount: int = 1) -> Union[int, None]:"""对指定键的值进行自增:param key: 键:param amount: 增量，默认为 1:return: 增加后的值，如果操作失败，则返回 None"""try:return self.redis_client.incrby(key, amount)except Exception as e:print(f"Error incrementing key {key}: {e}")return Nonedef decrement(self, key: str, amount: int = 1) -> Union[int, None]:"""对指定键的值进行自减:param key: 键:param amount: 减量，默认为 1:return: 减少后的值，如果操作失败，则返回 None"""try:return self.redis_client.decrby(key, amount)except Exception as e:print(f"Error decrementing key {key}: {e}")return Nonedef hset(self, hash_name: str, key: str, value: Any) -> bool:"""向哈希表中设置字段值:param hash_name: 哈希表名称:param key: 字段名:param value: 字段值:return: 是否成功"""try:self.redis_client.hset(hash_name, key, value)return Trueexcept Exception as e:print(f"Error setting hash {hash_name}:{key}: {e}")return Falsedef hget(self, hash_name: str, key: str) -> Optional[Any]:"""获取哈希表中的字段值:param hash_name: 哈希表名称:param key: 字段名:return: 字段值，如果字段不存在，则返回 None"""try:return self.redis_client.hget(hash_name, key)except Exception as e:print(f"Error getting hash {hash_name}:{key}: {e}")return Nonedef hgetall(self, hash_name: str) -> dict:"""获取哈希表中的所有字段和值:param hash_name: 哈希表名称:return: 键值对字典"""try:return self.redis_client.hgetall(hash_name)except Exception as e:print(f"Error getting all hash fields for {hash_name}: {e}")return {}# 示例使用
if __name__ == '__main__':redis_helper = RedisHelper()# 设置单个键值对redis_helper.set('name', 'Allen')# 获取单个键值print(redis_helper.get('name')) # Allen# 设置多个键值对redis_helper.set_multiple({'age': 18, 'location': 'shanghai'})# 获取多个键值print(redis_helper.get_multiple(['name', 'age', 'location'])) # {'name': 'Allen', 'age': '18', 'location': 'shanghai'}# 增加某个键的值redis_helper.increment('age')print(redis_helper.get('age')) # 19# 设置哈希表redis_helper.hset('user:1000', 'name', 'Allen')print(redis_helper.hget('user:1000', 'name')) # Allen