python-redis-module-skill

Python Redis Module Skill

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

能力清单

触发场景

用户说"帮我加 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()