95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
PerToolBox Server - Redis 客户端
|
|
Copyright (C) 2024 Sea Network Technology Studio
|
|
Author: Canglan <admin@sea-studio.top>
|
|
License: AGPL v3
|
|
"""
|
|
|
|
import redis
|
|
import json
|
|
import logging
|
|
from typing import Optional, Any
|
|
from ..config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class RedisClient:
|
|
def __init__(self):
|
|
self.client = None
|
|
self._connect()
|
|
|
|
def _connect(self):
|
|
try:
|
|
self.client = redis.Redis(
|
|
host=settings.REDIS_HOST,
|
|
port=settings.REDIS_PORT,
|
|
password=settings.REDIS_PASSWORD if settings.REDIS_PASSWORD else None,
|
|
db=settings.REDIS_DB,
|
|
decode_responses=True,
|
|
socket_timeout=5
|
|
)
|
|
self.client.ping()
|
|
logger.info("Redis 连接成功")
|
|
except Exception as e:
|
|
logger.error(f"Redis 连接失败: {e}")
|
|
self.client = None
|
|
|
|
def get(self, key: str) -> Optional[str]:
|
|
if not self.client:
|
|
return None
|
|
try:
|
|
return self.client.get(key)
|
|
except Exception as e:
|
|
logger.error(f"Redis get error: {e}")
|
|
return None
|
|
|
|
def set(self, key: str, value: Any, expire: int = None) -> bool:
|
|
if not self.client:
|
|
return False
|
|
try:
|
|
self.client.set(key, value, ex=expire)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Redis set error: {e}")
|
|
return False
|
|
|
|
def incr(self, key: str) -> int:
|
|
if not self.client:
|
|
return 0
|
|
try:
|
|
return self.client.incr(key)
|
|
except Exception as e:
|
|
logger.error(f"Redis incr error: {e}")
|
|
return 0
|
|
|
|
def delete(self, key: str) -> bool:
|
|
if not self.client:
|
|
return False
|
|
try:
|
|
self.client.delete(key)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Redis delete error: {e}")
|
|
return False
|
|
|
|
def exists(self, key: str) -> bool:
|
|
if not self.client:
|
|
return False
|
|
try:
|
|
return self.client.exists(key) > 0
|
|
except Exception as e:
|
|
logger.error(f"Redis exists error: {e}")
|
|
return False
|
|
|
|
def expire(self, key: str, seconds: int) -> bool:
|
|
if not self.client:
|
|
return False
|
|
try:
|
|
return self.client.expire(key, seconds)
|
|
except Exception as e:
|
|
logger.error(f"Redis expire error: {e}")
|
|
return False
|
|
|
|
redis_client = RedisClient() |