xy-utilities-redis 1.0.1

小阳工具库Redis模块 — Redis便捷操作封装

xy-utilities-redis

Redis Python 工具库，提供字符串、列表、集合、有序集合、哈希表等数据结构的简洁操作封装。

安装

pip install xy_utilities_redis

快速开始

from xy_utilities_redis import RedisHelper

# 推荐使用上下文管理器
with RedisHelper(host='127.0.0.1', port=6379) as redis:
    redis.string_set('key', 'value')
    print(redis.string_get('key'))  # 'value'

使用示例

String

redis.string_set('counter', 0)
redis.increase('counter')        # 1
redis.decrease('counter', 5)     # -4
value = redis.string_get('key')

List（队列）

redis.enqueue('queue', 'item1', 'item2')
item = redis.dequeue('queue')                    # 'item1'
items = redis.peek_range('queue', start=0, end=-1)  # 不出队，读取全部
length = redis.llen('queue')

Set

redis.set_add('myset', 'a', 'b', 'c')
redis.set_remove('myset', 'b')
members = redis.set_members('myset')  # ['a', 'c']

Sorted Set

redis.zset_add('ranking', {'Alice': 100, 'Bob': 90})
redis.zset_remove('ranking', 'Bob')
top = redis.zset_range('ranking', 0, -1)
by_score = redis.zset_range_byscore('ranking', min_score=80, max_score=100)
score = redis.zset_score('ranking', 'Alice')
redis.zset_remove_range('ranking', min_score=0, max_score=50)

Hash

redis.hash_set('user:1', {'name': 'Alice', 'age': '30'})
values = redis.hash_get('user:1', 'name', 'age')  # ['Alice', '30']
redis.hash_del('user:1', 'age')

Key 管理

keys = redis.key_get('user:*')
exists = redis.key_exists('user:1', 'user:2')  # 返回存在的 key 数量
redis.key_expire('session', 3600)              # 设置过期时间（秒）
ttl = redis.key_expiration('session')
redis.key_delete('session', 'tmp')

Pub/Sub

# 发布
redis.publish('channel', 'message')

# 订阅（handler 为回调函数）
pubsub = redis.subscribe('channel', lambda msg: print(msg['data']))
while True:
    pubsub.get_message()

API 参考

RedisHelper(**kwargs)

构造参数透传给 redis.StrictRedis。

分类	方法	说明
String	string_set(name, value)	设置字符串值
String	string_get(name) → str	获取字符串值
String	increase(name, amount=1) → int	自增
String	decrease(name, amount=1) → int	自减
List	enqueue(name, *values) → int	入队（RPUSH）
List	dequeue(name, count) → str|list	出队（LPOP）
List	peek_range(name, start, end) → list	读取不出队
List	lindex(name, index) → str	按下标读取
List	llen(name) → int	列表长度
Set	set_add(name, *values) → int	添加成员
Set	set_remove(name, *values) → int	移除成员
Set	set_members(name) → list	获取所有成员
ZSet	zset_add(name, mapping) → int	添加成员
ZSet	zset_remove(name, *values) → int	移除成员
ZSet	zset_range(name, start, end) → list	按排名范围获取
ZSet	zset_range_byscore(name, min_score, max_score) → list	按分数范围获取
ZSet	zset_score(name, member) → float	获取成员分数
ZSet	zset_remove_range(name, min_score, max_score) → int	按分数范围删除
Hash	hash_set(name, mapping) → int	批量设置字段
Hash	hash_get(name, *keys) → list	获取字段值
Hash	hash_del(name, *keys) → int	删除字段
Key	key_get(pattern) → list	按 pattern 搜索 key
Key	key_exists(*names) → int	检查 key 是否存在
Key	key_type(name) → str	获取 key 的值类型
Key	key_expire(name, time) → bool	设置过期时间
Key	key_expiration(name) → int	获取剩余 TTL
Key	key_delete(*names) → int	删除 key
Pub/Sub	publish(channel, message) → int	发布消息
Pub/Sub	subscribe(channel, handler) → PubSub	订阅频道

开发环境

conda create -n xy_utilities_redis python=3.11 -y
conda activate xy_utilities_redis
pip install build twine
pip install "redis>=6.2.0"

发布

rm -rf dist *.egg-info
python -m build

# 发布到私有 PyPI
twine upload -r packages-pypi dist/*

# 发布到 PyPI
python -m twine upload dist/*

License

MIT © Colin Chang