网站推广话术/seo排名优化公司哪家好

---

redis_decorator

安装

pip install redis_decorator

查看源代码

from io import StringIO
import pandas as pd
import json
import functoolsclass Cache:def __init__(self, redis_instance):if type(redis_instance.echo("hello")) == str:self.cache_container = redis_instanceelse:raise AttributeError("Redis instance's decode_responses must be set True. Use StrictRedis(..., decode_responses=True)")def key_generator(self, func, *args, **kwargs):return ":".join(["redis_dec", str(":".join([func.__name__, *[str(i) for i in args], str(kwargs)]))])def ttl(self, ttl=None, force_refresh=False):def enable(func):@functools.wraps(func)def func_wrapper(*args, **kwargs):target_key = self.key_generator(func, *args, **kwargs)a = self.cache_container.get(target_key)if a:return aelse:result = func(*args, **kwargs)self.cache_container.set(target_key, result, ttl)return resultreturn func_wrapperreturn enabledef delete_cache(self, func=None, *args, **kwargs):if func is None:print("Delete all the redis_dec")key = self.cache_container.scan(match="redis_dec:*")[1]elif not args and not kwargs:print("Remove every result related to this function")key = self.cache_container.scan(match=":".join(["redis_dec", func.__name__, "*"]))[1]else:key = [self.key_generator(func, *args, **kwargs)]if key:self.cache_container.delete(*key)return 0def _ser_df(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return func(*args, **kwargs).to_csv()return func_wrapperdef _de_ser_df(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):tmp = pd.read_csv(StringIO(func(*args, **kwargs)))return tmp.set_index(tmp.columns[0])return func_wrapperdef df(self, ttl=None):def deco(func):for dec in [self._ser_df, self.ttl(ttl), self._de_ser_df]:func = dec(func)return funcreturn decodef _ser_number(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return str(func(*args, **kwargs))return func_wrapperdef _de_ser_int(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return int(func(*args, **kwargs))return func_wrapperdef _de_ser_float(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return float(func(*args, **kwargs))return func_wrapperdef int(self, ttl=None):def deco(func):for dec in [self._ser_number, self.ttl(ttl), self._de_ser_int]:func = dec(func)return funcreturn decodef float(self, ttl=None):def deco(func):for dec in [self._ser_number, self.ttl(ttl), self._de_ser_float]:func = dec(func)return funcreturn decodef _ser_dict(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return json.dumps(func(*args, **kwargs))return func_wrapperdef _de_ser_dict(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return json.loads(func(*args, **kwargs))return func_wrapperdef dict(self, ttl=None):def deco(func):for dec in [self._ser_dict, self.ttl(ttl), self._de_ser_dict]:func = dec(func)return funcreturn decodef list(self, ttl=None):def deco(func):for dec in [self._ser_dict, self.ttl(ttl), self._de_ser_dict]:func = dec(func)return funcreturn decodef _de_ser_json(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return json.loads(func(*args, **kwargs))return func_wrapperdef _ser_json(self, func):@functools.wraps(func)def func_wrapper(*args, **kwargs):return json.dumps(json.loads(func(*args, **kwargs)))return func_wrapperdef json(self, ttl=None):def deco(func):for dec in [self._ser_json, self.ttl(ttl), self._de_ser_json]:func = dec(func)return funcreturn decoif __name__ == '__main__':pass

• Cache类：缓存
  • __init__：初始化，必须使用StrictRedis、参数decode_responses=True
  • ttl：设置过期时间，单位s
  • key_generator：生成key，没填写时，生成redis_dec:方法名:参数1:参数2…形式的key
  • delete_cache：删除缓存，删除全部缓存、指定函数缓存、指定函数及参数缓存
  • df：缓存df类型数据
  • int: 缓存int类型数据
  • float: 缓存float类型数据
  • dict：缓存dict类型数据
  • list ：缓存list类型数据
  • json：缓存json类型数据，和dict一样

使用

测试缓存int值

python -m unittest test.Test.test_my_int_function

测试批量删除

python -m unittest test.Test.test_delete

测试key生成与删除

python -m unittest test.Test.test_gen_key_delete

redis_decorators

安装

这里先安装一个其他人写的，体验一下

pip install redis_decorators

Successfully installed async-timeout-4.0.3 importlib-metadata-6.7.0 redis-5.0.3 redis-decorators-1.0.1 typing-extensions-4.7.1 zipp-3.15.0

注意：使用时如果报错，需要先删除__init__.py的12行

查看源代码

__init__.py

from .cache_element import CacheDateTime, CacheElement, CacheElementSingleType
from .cacheable import (Cacheable,DictCacheable,DictCacheType,DictStringCacheable,ListCacheable,ListCacheType,StringCacheable
)
from .caching import RedisCaching, build_redis_url

初始化，导入各种包

cacheable.py 各种可缓存的类型

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Generic, List, Optional, TypeVarfrom redis import RedisStoreType = TypeVar('StoreType')
ListCacheType = List[str]
DictCacheType = Dict[str, str]class Cacheable(Generic[StoreType], ABC):"""Performs caching store and fetch operations for a specific type.Subclass to define how to handle a specific type."""@abstractmethoddef store(self, client: Redis, key: str, value: StoreType) -> None:"""Store a value in cache.Args:client (Redis): Cache to store in.key (str): Name of cache value.value (StoreType): Value to store.Returns:None"""pass  # pragma: nocover@abstractmethoddef fetch(self, client: Redis, key: str) -> Optional[StoreType]:"""Fetch a value from cache.Args:client (Redis): Cache to fetch from.key (str): Name of cache value.Returns:StoreType or None: Value fetched from cache or None if no value exists."""pass  # pragma: nocoverclass StringCacheable(Cacheable[str]):def store(self, client: Redis, key: str, value: str):client.set(key, value)def fetch(self, client: Redis, key: str) -> Optional[str]:return client.get(key)@dataclass
class DictStringCacheable(Cacheable[str]):"""Attributes:dict_key (str): Name of hash value."""dict_key: strdef store(self, client: Redis, key: str, value: str):client.hset(key, self.dict_key, value)def fetch(self, client: Redis, key: str) -> Optional[str]:return client.hget(key, self.dict_key)class DictCacheable(Cacheable[DictCacheType]):def store(self, client: Redis, key: str, value: DictCacheType):client.hset(key, mapping=value)def fetch(self, client: Redis, key: str) -> Optional[DictCacheType]:return client.hgetall(key) or Noneclass ListCacheable(Cacheable[ListCacheType]):def store(self, client: Redis, key: str, value: ListCacheType):client.delete(key)client.rpush(key, *value)def fetch(self, client: Redis, key: str) -> Optional[ListCacheType]:return client.lrange(key, 0, -1) or None

• Cacheable类：抽象类，不可实例化。对特定类型执行缓存存储（store）和提取（fetch）操作。子类来定义如何处理特定类型。
• StringCacheable类：Cacheable子类，使用set、get处理字符串。
• DictStringCacheable类：Cacheable子类，使用hset、hget处理字典，有dict_key，有dataclass装饰，将会自动添加__init__等方法。
• DictCacheable类：Cacheable子类，使用hset、hgetall处理字典。
• ListCacheable类：Cacheable子类，使用delete、rpush、lrange处理列表。

cache_element.py 缓存的元素

from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Generic, Optional, TypeVarfrom redis import Redisfrom .cacheable import Cacheable, StringCacheableFetchType = TypeVar('FetchType')
StoreType = TypeVar('StoreType')class CacheElement(Generic[FetchType, StoreType], ABC):"""Get and set cache values.Attributes:cacheable (Cacheable): Instance used to store and fetch values."""cacheable: Cacheable[StoreType]def get_value(self, client: Redis, key: str) -> Optional[FetchType]:"""Returns cached value or None if no value exists."""value = self.cacheable.fetch(client, key)if value is None:return Nonereturn self.load(value)def set_value(self, client: Redis, key: str, value: FetchType) -> None:"""Set value in cache.Args:client (Redis): Cache to fetch from.key (str): Name of cache value.Returns:None"""self.cacheable.store(client, key, self.dump(value))@abstractmethoddef load(self, value: StoreType) -> FetchType:"""Load value from cache into expected Python type."""pass  # pragma: nocover@abstractmethoddef dump(self, value: FetchType) -> StoreType:"""Dump value from Python type into type expected by cache."""pass  # pragma: nocover@dataclass
class CacheElementSingleType(CacheElement[FetchType, FetchType]):"""A CacheElement that fetches the same type that it stores.By default, values are passed to and from cache as-is, i.e. no serializationor deserialization is performed.For example, a string can be stored and fetched without modification whereasa datetime would need to be serialized for storage and deserialized forretrieval."""cacheable: Cacheable[FetchType]def load(self, value: FetchType) -> FetchType:return valuedef dump(self, value: FetchType) -> FetchType:return value@dataclass
class CacheDateTime(CacheElement[datetime, str]):"""Store and fetch datetime values with string serialization."""cacheable: Cacheable[str] = StringCacheable()def dump(self, value: datetime) -> str:return value.isoformat()def load(self, value: str) -> datetime:return datetime.fromisoformat(value)

• CacheElement类：抽象类。缓存元素，子类实现设置和获取缓存值。
• CacheElementSingleType类：CacheElement子类，有cacheable，有dataclass装饰，自动添加__init__等方法。
• CacheDateTime类：CacheElement子类，通过字符串序列化的方式缓存日期类型，有cacheable，有dataclass装饰，自动添加__init__等方法。

caching.py 缓存主要逻辑

from datetime import timedelta
from functools import partial, update_wrapper
from typing import Any, Callable, Optional, Unionfrom redis import Redisfrom .cache_element import CacheDateTime, CacheElement, CacheElementSingleType
from .cacheable import (DictCacheable,DictCacheType,DictStringCacheable,ListCacheable,ListCacheType,StringCacheable
)class RedisCaching:"""Provides decorators for automatic caching."""cache_cls = Redis_cache_instances = {}def __init__(self, url=None, **kwargs):self._default_cache_kwargs = {'decode_responses': True, 'socket_timeout': 15}self.init(url, **kwargs)def init(self, url, **kwargs):self._url = urlself._cache_kwargs = {**self._default_cache_kwargs,**kwargs,}def get_cache(self) -> Redis:if self._url in self._cache_instances:return self._cache_instances.get(self._url)cache = self.cache_cls.from_url(self._url, **self._cache_kwargs)self._cache_instances[self._url] = cachereturn cachedef delete(self, cache_key: str):self.get_cache().delete(cache_key)def cache_value(self,cache_element: CacheElement,get_cache_key: Callable[..., str] = None,expire_in: Union[int, timedelta] = None,):"""Decorate a function to automatically cache its return value.Wrapper does two things:1. If cached value does not exist, cache the return value of the function.2. If cached value exists, return it instead of calling the function.Args:cache_element (CacheElement): Instance used to get and set cache value.get_cache_key (Callable): Function that returns name of cache value.Accepts the same arguments as the decorated function.expire_in (Union[int, timedelta]): Number of seconds until this keyexpires after being set. Can be a datetime.timedelta object.Examples:Decorate a function that returns a string:.. code-block:: [e-mail]_string(get_cache_key=lambda arg: f'object:{arg}')def expensive_fetch_operation(arg) -> str:...return computed_valueUse `cache_key` of decorated function to set `get_cache_key`:.. code-block:: [e-mail]_string()def expensive_fetch_operation(arg) -> str:...return computed_value@expensive_fetch_operation.cache_keydef expensive_fetch_operation_cache_key(arg) -> str:...return computed_value"""def decorator(func):return CacheValueWrapper(self, func, cache_element, get_cache_key, expire_in)return decoratordef cache_string(self, get_cache_key: Callable[..., str] = None, **kwargs):"""Decorate a function to store a string."""return self.cache_value(CacheElementSingleType[str](cacheable=StringCacheable()),get_cache_key,**kwargs,)def cache_dict(self, get_cache_key: Callable[..., str] = None, **kwargs):"""Decorate a function to store a dictionary {str: str}."""return self.cache_value(CacheElementSingleType[DictCacheType](cacheable=DictCacheable()),get_cache_key,**kwargs,)def cache_dict_string(self, dict_key: str, get_cache_key=None, **kwargs):"""Decorate a function to store a specific key inside a cached hash."""return self.cache_value(CacheElementSingleType[str](cacheable=DictStringCacheable(dict_key)),get_cache_key,**kwargs,)def cache_list(self, get_cache_key: Callable[..., str] = None, **kwargs):"""Decorate a function to store a list of strings."""return self.cache_value(CacheElementSingleType[ListCacheType](cacheable=ListCacheable()),get_cache_key,**kwargs,)def cache_datetime(self, get_cache_key: Callable[..., str] = None, **kwargs):"""Decorate a function to store a datetime."""return self.cache_value(CacheDateTime(), get_cache_key, **kwargs)class CacheValueWrapper:def __init__(self,caching: RedisCaching,func: Callable,cache_element: CacheElement,get_cache_key: Optional[Callable[..., str]] = None,expire_in: Union[int, timedelta] = None,):self._caching = cachingself._func = funcself._cache_element = cache_elementself._get_cache_key = get_cache_keyself._expire_in = expire_inupdate_wrapper(self, func)def __call__(self, *args: Any, **kwargs: Any):cache_key = self._calculate_cache_key(*args, **kwargs)value = self._cache_element.get_value(self._caching.get_cache(), cache_key)if value is None:client = self._caching.get_cache()value = self._func(*args, **kwargs)self._cache_element.set_value(client, cache_key, value)expire_in = self._calculate_expire_in(value, *args, **kwargs)if expire_in:client.expire(cache_key, expire_in)return valuedef __get__(self, instance, owner):return partial(self, instance)def cache_key(self, func):self._get_cache_key = funcreturn funcdef expire_in(self, func):self._expire_in = funcreturn funcdef _calculate_expire_in(self, value, *args, **kwargs):if callable(self._expire_in):kwargs['value'] = valuereturn self._expire_in(*args, **kwargs)return self._expire_indef _calculate_cache_key(self, *args: Any, **kwargs: Any):if self._get_cache_key is None:arg_str = ':'.join([self._func.__name__, *[str(arg) for arg in args], str(kwargs)])return ':'.join(['redis_decorators', arg_str])return self._get_cache_key(*args, **kwargs)def build_redis_url(host, password, db, use_secure=True):prefix = 'rediss' if use_secure else 'redis'if password:url = f'{prefix}://:{password}@{host}'else:url = f'{prefix}://{host}'if db:url = f'{url}/{db}'return url

• RedisCaching类：提供自动化缓存装饰器

  • get_cache方法：获取Redis实例
  • delete方法：删除指定key
  • cache_value：装饰器，自动缓存函数返回值
  • cache_string：缓存字符串
  • cache_dict：缓存字典
  • cache_dict_string：
  • cache_list：缓存列表
  • cache_datetime：缓存日期时间

• CacheValueWrapper类：

  • cache_key：缓存key
  • expire_in：过期时间
  • _calculate_cache_key：如果cache_key为None时，计算key为redis_decorators:函数名:参数1:参数2…
  • _calculate_expire_in：计算过期时间

• build_redis_url：根据参数构建url

使用

缓存字符串

python -m unittest test.Test.test_my_string_function

python -m unittest test.Test.test_my_datetime_function

python -m unittest test.Test.test_my_list_function

python -m unittest test.Test.test_my_dict_function

测试删除

python -m unittest test.Test.test_delete_key

总结

redis缓存装饰器需要考虑以下几点：

1. key生成：用户不指定key时需要自动生成，统一前缀方便后序更新、删除
2. 指定key：用户可以指定key（redis_decorator无法指定key）
3. 缓存和获取功能
4. 常见类型的缓存支持
5. 用户自定义类型可以继承抽象类，实现序列化和反序列化方法（redis_decorator没有）
6. 指定key的更新（包没有）
7. 指定或批量key的删除（redis_decorators只有删除指定key）
8. 支持协程（包没有）

全部代码

main.py

from redis import StrictRedis
from redis_dec import Cache
from redis_decorators import RedisCaching
from datetime import datetime
from settings import REDIS_URLrediss = StrictRedis.from_url(REDIS_URL,decode_responses=True)
redis_cache = Cache(rediss)caching = RedisCaching(REDIS_URL)
cache = caching.get_cache()  # Redis instance@redis_cache.int()
def my_int_function(a: int, b: int):print(f"my_int_function excuting with {a} {b}...")return [e-mail]_string()
def my_string_function(arg1, arg2):print(f"my_string_function excuting with {arg1} {arg2}...")return str(arg1+arg2)@caching.cache_datetime()
def my_datetime_function(date):print(f"my_date_function excuting with {date}")return datetime.now()@caching.cache_list()
def my_list_function(*arg):print(f"my_list_function excuting with {arg}")return list(arg)@caching.cache_dict()
def my_dict_function(data:dict):print(f"my_dict_function excuting with {data}")return data

test.py

import unittestclass Test(unittest.TestCase):def test_gen_key_delete(self):from main import redis_cachefrom main import my_int_functionkey = redis_cache.key_generator(my_int_function,5,6)redis_cache.cache_container.delete(key)def test_my_int_function(self):from main import my_int_functionres = my_int_function(5,6)print(res)def test_delete(self):from main import redis_cachefrom main import my_int_functionres = redis_cache.delete_cache(func=my_int_function)print(res)def test_my_string_function(self):from main import my_string_functionres = my_string_function(arg1=1,arg2=2)print(res)def test_my_datetime_function(self):from main import my_datetime_functionres = my_datetime_function("now")print(res)def test_my_list_function(self):from main import my_list_functionres = my_list_function(4,5,6)print(res)def test_my_dict_function(self):from main import my_dict_functionres = my_dict_function({"1":1})print(res)def test_delete_key(self):from main import cachingfrom typing import Anyfunc_name = "my_datetime_function"def _calculate_cache_key(*args: Any, **kwargs: Any):arg_str = ':'.join([func_name, *[str(arg) for arg in args], str(kwargs)])return ':'.join(['redis_decorators', arg_str])caching.delete(_calculate_cache_key("now"))

参考

redis-python客户端官方文档

pypi-redis-decorator

python3-abc包
python3-dataclasses包
pypi-redis-decorators
github-python-redis-decorators