mirror of https://gitee.com/anolis/sysom.git
98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
# -*- coding: utf-8 -*- #
|
|
"""
|
|
Time 2023/4/28 15:07
|
|
Author: mingfeng (SunnyQjm)
|
|
Email mfeng@linux.alibaba.com
|
|
File redis_gcache.py
|
|
Description:
|
|
"""
|
|
import json
|
|
from typing import Union, Dict
|
|
from gcache_base import GCache, GCacheUrl, GCacheException
|
|
from redis_lua import XRedisHashTable
|
|
from .common import ClientBase, StaticConst
|
|
|
|
SEPARATOR = "_GCache_"
|
|
|
|
|
|
class RedisGCache(GCache, ClientBase):
|
|
"""
|
|
A redis-based gcache implement
|
|
"""
|
|
|
|
def __init__(self, cache_name, url: GCacheUrl):
|
|
GCache.__init__(self)
|
|
ClientBase.__init__(self, url)
|
|
self._x_redis_hash_table = XRedisHashTable(self.redis_client)
|
|
self._table_name = cache_name
|
|
|
|
@staticmethod
|
|
def _to_str(value: Union[int, float, dict, str]):
|
|
if isinstance(value, int):
|
|
return str(value)
|
|
elif isinstance(value, float):
|
|
return str(value)
|
|
elif isinstance(value, str):
|
|
return value
|
|
elif isinstance(value, dict):
|
|
return json.dumps(value)
|
|
|
|
@staticmethod
|
|
def _get_store_value(value: Union[int, float, dict, str]):
|
|
return f"{type(value).__name__}{SEPARATOR}{RedisGCache._to_str(value)}"
|
|
|
|
@staticmethod
|
|
def _from_str(type_v: str, value_str: str) -> Union[int, float, dict, str]:
|
|
if type_v == "int":
|
|
return int(value_str)
|
|
elif type_v == "float":
|
|
return float(value_str)
|
|
elif type_v == "str":
|
|
return value_str
|
|
elif type_v == "dict":
|
|
return json.loads(value_str)
|
|
|
|
def store(self, key: str, value: Union[int, float, dict, str],
|
|
expire: int = -1) -> bool:
|
|
return self._x_redis_hash_table.hset(
|
|
self._table_name, key,
|
|
f"{self._get_store_value(value)}",
|
|
expire=expire
|
|
)
|
|
|
|
def _get_format_value(self, value: str) -> Union[None, int, float, dict, str]:
|
|
type_value = value.split(SEPARATOR)
|
|
if len(type_value) < 2:
|
|
raise GCacheException(
|
|
f"Load failed, expect value which is {value} start "
|
|
f"with {SEPARATOR}, "
|
|
)
|
|
type_v = type_value[0]
|
|
value = "".join(type_value[1:])
|
|
if type_v not in ["int", "float", "dict", "str"]:
|
|
raise GCacheException(
|
|
f"Got not supported type = {type_v}, expect one of "
|
|
f"[int, float, dict, str]"
|
|
)
|
|
return self._from_str(type_v, value)
|
|
|
|
def load(self, key: str) -> Union[None, int, float, dict, str]:
|
|
res = self._x_redis_hash_table.hget(self._table_name, key)
|
|
if res is None:
|
|
return None
|
|
return self._get_format_value(res)
|
|
|
|
def load_all(self) -> Dict[str, Union[int, float, dict, str]]:
|
|
res = self._x_redis_hash_table.hgetall(self._table_name)
|
|
if res is None:
|
|
return {}
|
|
for k in res:
|
|
res[k] = self._get_format_value(res[k])
|
|
return res
|
|
|
|
def clean(self):
|
|
self._x_redis_hash_table.hdrop_table(self._table_name)
|
|
|
|
def delete(self, key: str) -> bool:
|
|
return self._x_redis_hash_table.hdel(self._table_name, key)
|