接口自动化测试之Redis工具类的封装
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import redis
class RedisConfig:
"""
Redis 配置信息
"""
def __init__(self, host='127.0.0.1', port=6379, password=None, db=0, max_connections=10):
self.host = host
self.port = port
self.password = password
self.db = db
self.max_connections = max_connections
class RedisPool:
"""
Redis 连接池
"""
def __init__(self, config):
"""
初始化 Redis 连接池
:param config: RedisConfig 实例
"""
self.pool = redis.ConnectionPool(
host=config.host,
port=config.port,
password=config.password,
db=config.db,
max_connections=config.max_connections,
socket_timeout=5,
socket_connect_timeout=2,
decode_responses=True,
retry_on_timeout=True
)
def get_connection(self):
"""
获取一个 Redis 连接对象
:return: Redis 连接对象
"""
return redis.StrictRedis(connection_pool=self.pool)
def __enter__(self):
"""
进入上下文管理器时返回一个 Redis 连接对象
"""
return self.get_connection()
def __exit__(self, exc_type, exc_val, exc_tb):
"""
退出上下文管理器时自动关闭 Redis 连接
"""
self.pool.disconnect()
class RedisClient:
"""
Redis 工具类
"""
def __init__(self, config):
self.config = config
self.pool = RedisPool(config)
def get(self, key):
"""
获取指定 key 的值
:param key: 要获取的 key
:return: key 对应的 value，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.get(key)
except redis.RedisError:
return None
def set(self, key, value, ex=None, px=None, nx=False, xx=False):
"""
设置指定 key 的值
:param key: 要设置的 key
:param value: key 对应的值
:param ex: 过期时间，单位为秒
:param px: 过期时间，单位为毫秒
:param nx: 如果key不存在才添加，True:不存在可以添加，False:不存在不可添加
:param xx: 如果key已经存在才添加，True:存在可以添加，False:存在不可添加
:return: 是否设置成功（True/False）
"""
try:
with self.pool.get_connection() as conn:
return conn.set(key, value, ex=ex, px=px, nx=nx, xx=xx)
except redis.RedisError:
return False
def incr(self, key, amount=1):
"""
将指定 key 的值增加指定的数量，默认增加 1
:param key: 要增加值的 key
:param amount: 要增加的数量
:return: 增加后的值，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.incr(key, amount)
except redis.RedisError:
return None
def decr(self, key, amount=1):
"""
将指定 key 的值减少指定的数量，默认减少 1
:param key: 要减少值的 key
:param amount: 要减少的数量
:return: 减少后的值，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.decr(key, amount)
except redis.RedisError:
return None
def delete(self, key):
"""
删除指定的 key
:param key: 要删除的 key
:return: 是否删除成功（True/False）
"""
try:
with self.pool.get_connection() as conn:
return conn.delete(key)
except redis.RedisError:
return False
def hget(self, name, key):
"""
获取指定哈希表中指定字段的值
:param name: 哈希表的名称
:param key: 哈希表中的字段名
:return: 哈希表中的值，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.hget(name, key)
except redis.RedisError:
return None
def hset(self, name, key, value):
"""
设置指定哈希表中指定字段的值
:param name: 哈希表的名称
:param key: 哈希表中的字段名
:param value: 要设置的值
:return: 是否设置成功（True/False）
"""
try:
with self.pool.get_connection() as conn:
return conn.hset(name, key, value)
except redis.RedisError:
return False
def hdel(self, name, key):
"""
删除指定哈希表中的指定字段
:param name: 哈希表的名称
:param key: 哈希表中的字段名
:return: 是否删除成功（True/False）
"""
try:
with self.pool.get_connection() as conn:
return conn.hdel(name, key)
except redis.RedisError:
return False
def lpush(self, key, *values):
"""
将一个或多个值插入到列表头部
:param key: 要插入值的列表名
:param values: 要插入的一个或多个值
:return: 列表当前的长度，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.lpush(key, *values)
except redis.RedisError:
return None
def rpop(self, key):
"""
移除并返回列表的最后一个元素
:param key: 列表名
:return: 列表的最后一个元素，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.rpop(key)
except redis.RedisError:
return None
def smembers(self, name):
"""
获取集合中的所有成员
:param name: 集合的名称
:return: 集合中的所有成员，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.smembers(name)
except redis.RedisError:
return None
def sadd(self, name, *values):
"""
向集合中添加一个或多个成员
:param name: 集合的名称
:param values: 要添加的一个或多个成员
:return: 添加成功的成员数，不包括已存在的成员，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.sadd(name, *values)
except redis.RedisError:
return None
def srem(self, name, *values):
"""
从集合中移除一个或多个成员
:param name: 集合的名称
:param values: 要移除的一个或多个成员
:return: 移除成功的成员数，不包括不存在的成员，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.srem(name, *values)
except redis.RedisError:
return None
def get_all_keys(self, pattern='*'):
"""
获取所有符合 pattern 的 key 列表
:param pattern: 匹配模式，默认为 *
:return: 符合 pattern 的 key 列表，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.keys(pattern)
except redis.RedisError:
return None
def get_hash_all(self, name):
"""
获取指定哈希表的所有字段与值
:param name: 哈希表名称
:return: 哈希表中的所有字段与值，失败返回 None
"""
try:
with self.pool.get_connection() as conn:
return conn.hgetall(name)
except redis.RedisError:
return None
def set_hash_all(self, name, data_dict):
"""
设置指定哈希表的所有字段与值
:param name: 哈希表名称
:param data_dict: 字典类型的数据，需要设置的所有字段与值
:return: 是否设置成功（True/False）
"""
try:
with self.pool.get_connection() as conn:
for k, v in data_dict.items():
conn.hset(name, k, v)
return True
except redis.RedisError:
return False
def close(self):
"""
关闭连接，释放连接池占用的资源
"""
self.pool.__exit
if __name__ == '__main__':
redis_config = RedisConfig(host='localhost', port=6379, password=None, db=0, max_connections=10)
with RedisClient(redis_config) as redis_client:
# 使用 Redis 操作方法
redis_client.set('key1', 'value1')
value = redis_client.get('key1')
print(value)
以上代码是一个 Redis 工具类的实现，用于 Python 对 Redis 进行操作。其中包含三个主要类：RedisConfig、RedisPool 和 RedisClient。
RedisConfig 类用于存储 Redis 配置信息。在初始化时需要提供 Redis 的主机地址、端口号、密码、数据库编号、最大连接数等基本配置信息。这些信息用于创建 Redis 连接池所需的参数，并在 RedisPool 类中被使用。
RedisPool 类用于创建 Redis 连接池。在初始化时需要传入 RedisConfig 实例，根据该实例中的配置信息生成 Redis 连接池并保存在 self.pool 中。get_connection 方法从连接池中获取一个 Redis 连接对象，并返回该对象。enter 和 exit 方法分别用于进入和退出上下文管理器。在进入时返回一个 Redis 连接对象，在退出时关闭 Redis 连接。
RedisClient 类是 Redis 工具类的核心。在初始化时需要传入 RedisConfig 实例，同时将其传入 RedisPool 构造函数以创建连接池。该类中包括了 set、get、incr、decr 等常见操作和其他对 Redis 的一些常用操作。在每个方法中，首先调用 get_connection 方法获取一个 Redis 连接对象，然后通过该对象进行相应的 Redis 操作。在操作完成后释放该连接。
除了常见的操作方法外，RedisClient 还提供了一些特殊的操作方法，如哈希表操作（hget、hset、hdel）、列表操作（lpush、rpop）、集合操作（smembers、sadd、srem）以及获取所有符合指定 pattern 的 key 列表（get_all_keys）等。这些方法通过 Redis 连接对象实现，具体操作方法可查看 redis-py 的官方文档。
在使用 Redis 时，使用 with RedisClient(connection_pool) as redis_client 的方式创建 RedisClient 实例，并进行相应的操作。例如可以调用 set 方法来设置键值对，get 方法来获取指定 key 的值等。
总之，以上代码是一个对 Redis 进行操作的工具类，具有易于使用、高性能、灵活配置和安全性等优点，可用于 Python 项目中
评论