Skip to main content
Join the official 2019 Python Developers SurveyStart the survey!

No project description provided

Project description

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for red-cache, version 0.0.7
Filename, size File type Python version Upload date Hashes
Filename, size red_cache-0.0.7-py3-none-any.whl (10.7 kB) File type Wheel Python version py3 Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page