python-redis-module-skill

Python Redis Module Skill

面向已有 FastAPI 项目的开发者，快速集成 Redis 能力。

能力清单

能力	说明
缓存	装饰器缓存、缓存更新、缓存删除
Session	Redis Session 共享
分布式锁	基于 Redis 的分布式锁
限流	基于 Redis 的接口限流
消息队列	Redis Stream 消息发布/订阅
计数器	分布式计数器

触发场景

用户说"帮我加 Redis"或"集成 Redis"时触发。

依赖配置

pip install redis aioredis

默认方法封装

1. Redis 客户端

# redis_client.py
import redis
from typing import Optional

class RedisClient:
    def __init__(self, host: str = "localhost", port: int = 6379, password: Optional[str] = None, db: int = 0):
        self.client = redis.Redis(
            host=host,
            port=port,
            password=password,
            db=db,
            decode_responses=True
        )
    
    # 基础操作
    def get(self, key: str) -> Optional[str]:
        return self.client.get(key)
    
    def set(self, key: str, value: str, ex: Optional[int] = None) -> bool:
        return self.client.set(key, value, ex=ex)
    
    def delete(self, *keys: str) -> int:
        return self.client.delete(*keys)
    
    def exists(self, key: str) -> bool:
        return self.client.exists(key)
    
    def expire(self, key: str, seconds: int) -> bool:
        return self.client.expire(key, seconds)
    
    # 哈希操作
    def hget(self, name: str, key: str) -> Optional[str]:
        return self.client.hget(name, key)
    
    def hset(self, name: str, key: str, value: str) -> int:
        return self.client.hset(name, key, value)
    
    def hgetall(self, name: str) -> dict:
        return self.client.hgetall(name)
    
    # 列表操作
    def lpush(self, name: str, *values: str) -> int:
        return self.client.lpush(name, *values)
    
    def rpop(self, name: str) -> Optional[str]:
        return self.client.rpop(name)
    
    # 自增
    def incr(self, key: str, amount: int = 1) -> int:
        return self.client.incr(key, amount)
    
    # 分布式锁
    def lock(self, key: str, timeout: int = 30) -> bool:
        return self.client.set(f"lock:{key}", "1", nx=True, ex=timeout)
    
    def unlock(self, key: str) -> int:
        return self.client.delete(f"lock:{key}")

# 全局实例
redis_client = RedisClient()