mirror of https://gitee.com/anolis/sysom.git
94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
# -*- coding: utf-8 -*- #
|
|
"""
|
|
Time 2022/7/29 13:33
|
|
Author: mingfeng (SunnyQjm)
|
|
Email mfeng@linux.alibaba.com
|
|
File common.py
|
|
Description:
|
|
"""
|
|
from typing import Optional
|
|
from clogger import logger
|
|
from redis import Redis
|
|
from gcache_base import GCacheUrl
|
|
from .utils import do_connect_by_gcache_url
|
|
|
|
|
|
class StaticConst:
|
|
"""Static consts
|
|
|
|
This class defines all the static constant values in the cmg-redis module
|
|
"""
|
|
# List of specialization parameters
|
|
REDIS_SPECIAL_PARM_G_CACHE_TABLE_NAME = \
|
|
"table_name"
|
|
|
|
_redis_special_parameter_list = [
|
|
REDIS_SPECIAL_PARM_G_CACHE_TABLE_NAME
|
|
]
|
|
|
|
_redis_special_parameters_default_value = {
|
|
REDIS_SPECIAL_PARM_G_CACHE_TABLE_NAME: (str, True)
|
|
}
|
|
|
|
@staticmethod
|
|
def parse_special_parameter(params: dict) -> dict:
|
|
"""Parse specialization parameters
|
|
|
|
Parse the specialization parameters and remove the specialization
|
|
parameters from the parameter list
|
|
|
|
Args:
|
|
params(dict): CecUrl.params
|
|
|
|
Returns:
|
|
|
|
"""
|
|
res = {}
|
|
for key in StaticConst._redis_special_parameter_list:
|
|
_type, default = \
|
|
StaticConst._redis_special_parameters_default_value[key]
|
|
res[key] = _type(params.pop(key, default))
|
|
return res
|
|
|
|
|
|
class ClientBase:
|
|
"""
|
|
cec-redis client base class, Redis* requires inherit from this class,
|
|
which provides some generic implementation
|
|
"""
|
|
|
|
def __init__(self, url: GCacheUrl):
|
|
self._redis_version = None
|
|
self._special_params = StaticConst.parse_special_parameter(url.params)
|
|
self.redis_client: Optional[Redis] = None
|
|
self.current_url = ""
|
|
self.connect_by_gcache_url(url)
|
|
|
|
def get_special_param(self, key: str, default=''):
|
|
"""Get specialization parameter by key
|
|
|
|
Args:
|
|
key(str): specialization parameter key
|
|
default(Any): default value if key not exists
|
|
|
|
Returns:
|
|
|
|
"""
|
|
return self._special_params.get(key, default)
|
|
|
|
def connect_by_gcache_url(self, url: GCacheUrl):
|
|
"""Connect to redis server by CmgUrl
|
|
|
|
Connecting to the Redis server via CmgUrl
|
|
|
|
Args:
|
|
url(str): CmgUrl
|
|
"""
|
|
logger.debug(
|
|
f"{self} try to connect to '{url}'.")
|
|
self.redis_client = do_connect_by_gcache_url(url)
|
|
self.current_url = str(url)
|
|
logger.info(
|
|
f"{self} connect to '{url}' successfully.")
|
|
return self
|