67 lines
1.7 KiB
Python
67 lines
1.7 KiB
Python
"""
|
||
redis 配置类
|
||
"""
|
||
import sys
|
||
|
||
import redis
|
||
|
||
from configs import default
|
||
|
||
|
||
class RedisCli(object):
|
||
|
||
def __init__(self, *, host: str, port: str, password: str, db: int, socket_timeout=5):
|
||
# redis对象，在 @app.on_event('startup') 中连接创建
|
||
self._redis_client = None
|
||
self.host = host
|
||
self.port = port
|
||
self.password = password
|
||
self.db = db
|
||
self.socket_timeout = socket_timeout
|
||
|
||
def init_redis_connect(self) -> None:
|
||
"""
|
||
初始化链接
|
||
:return:
|
||
"""
|
||
try:
|
||
r = redis.ConnectionPool(
|
||
host=self.host,
|
||
port=self.port,
|
||
password=self.password,
|
||
db=self.db,
|
||
socket_timeout=self.socket_timeout,
|
||
decode_responses=True # 解码
|
||
)
|
||
self._redis_client = redis.StrictRedis(connection_pool=r)
|
||
if not self._redis_client.ping():
|
||
print('连接超时')
|
||
sys.exit()
|
||
except (redis.AuthenticationError, Exception) as e:
|
||
print('连接redis异常')
|
||
sys.exit()
|
||
|
||
def get_redis(self):
|
||
return self._redis_client
|
||
|
||
# 使实例化后的对象，赋予redis对象的方法和属性
|
||
def __getattr__(self, item):
|
||
return self._redis_client.has_key(item)
|
||
|
||
def __getitem__(self, item):
|
||
return self._redis_client[item]
|
||
|
||
def __setitem__(self, key, value):
|
||
self._redis_client[key] = value
|
||
|
||
def __delitem__(self, key):
|
||
del self._redis_client[key]
|
||
|
||
|
||
redis_client = RedisCli(
|
||
host=default.db['host'],
|
||
port=default.db['port'],
|
||
password=default.db['password'],
|
||
db=default.db['db'],
|
||
)
|