red_cache

介绍

基于Redis实现的Python缓存工具

示例

安装方法

$ pip install red-cache

缓存函数执行结果

@redis_cache.pickle_cache(key=lambda v, t: "cache:{}:{}".format(v, t), ex=180)
def hell_world(val: str, times: int):
    return val * times

保存JSON缓存数据

@redis_cache.json_cache(key=lambda v, t: "cache:{}:{}".format(v, t), ex=180)
def hell_world(val: str, times: int):
    return {"val": val, "times": times}

缓存的属性

class Demo:
    def load_xxx(self):
        return

    xxx = redis_cache.property(key=lambda: "Demo::xxx", ex=10)(lambda self: self.load_xxx())

删除缓存

@redis_cache.remove(lambda o: "auth::user:{}".format(o))
def modify_user(user_id):
    # DO MODIFY USER
    pass

使用返回值

@redis_cache.remove(lambda o: "auth::user:{}".format(o), by_return=True)
def modify_user(user_id):
    # DO MODIFY USER
    return "*********"

使用生成器

@redis_cache.remove(lambda o: "auth::user:{}".format(o), by_return=True)
def modify_users(users):
    # modify users
    for u in users:
        yield u

基于Redis的分布式锁

@red_cache.red_lock(lambda uid, **kwargs: "red::lock:{}".format(uid), ttl=100000, retry_times=10, retry_delay=200)
def modify_user(uid: str, **kwargs):
    # DO MODIFY USER
    pass

临时令牌工具

import uuid

from red_cache import RedisCache, CachedToken

red_cache = RedisCache(dict(host='10.0.0.11', db=9))


# 声明 Token令牌类，集成CachedToken
class Token(CachedToken, metaclass=red_cache.token_meta):

    # 使用metaclass时可自动注入RedisCache对象到当前类对象
    # 使用类属性`red_cache`指定绑定的RedisCache亦可

    def __init__(self, token: str, username: str):
        super().__init__()
        self.token = token
        self.username = username

    # 使用cache_key_prefix指定缓存名称前缀
    cache_key_prefix = property(lambda self: self.__class__.__name__)

    # id 即当前令牌对象唯一值
    @property
    def id(self):
        return self.token

    # 返回字典，CachedToken使用标准库的json包序列化该字典作为对应缓存的值
    def marshal(self) -> dict:
        return dict(token=self.token, username=self.username)

    @classmethod
    def new(cls, username: str):
        return cls(token=uuid.uuid1().hex, username=username)


if __name__ == '__main__':
    tk = Token.new('/**/').save()
    # 使用ID读取令牌
    tk = Token.read(tk.token)
    # 刷新，即强制写入令牌到Redis
    tk.flush()
    # 删除令牌
    tk.remove()

计数器工具

from red_cache import RedisCache

red_cache = RedisCache(dict(host='10.0.0.11'))


class A:
    # 自增计数器
    counter = red_cache.counter("resource::a", 3,init=lambda:20*20)
    # 自减计数器
    desc = red_cache.counter("resource:desc", -2)
    # 基于HASH的自增计数器
    score = red_cache.hash_counter("score", "a", 1,init=lambda:20+1)

    def __init__(self):
        self._sign = red_cache.counter("{}::sign:{}".format(self.__class__.__name__, id(self)))

    # 使用属性
    sign = property(lambda self: self._sign.get())

@author:Memory_Leak