38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from enum import Enum
|
|
|
|
from app.utils.redis_utils import redis_utils, remove_redis_cache
|
|
|
|
|
|
class ApiSecretStatus(Enum):
|
|
valid = "valid" # 秘钥有效
|
|
invalid = "invalid" # 秘钥无效
|
|
not_exist = "not_exist" # 秘钥不存在
|
|
|
|
|
|
class ApiSecretUtils():
|
|
def __init__(self):
|
|
self.cache = {}
|
|
|
|
# 验证秘钥状态
|
|
async def verify_secret(self, secret: str) -> ApiSecretStatus:
|
|
async with redis_utils.get_redis_connection() as redis_client:
|
|
cache_value = await redis_client.get(f"api_secret_status_${secret}")
|
|
print("cache_value", cache_value)
|
|
|
|
api_secret_status = cache_value
|
|
print("verify_secret", api_secret_status, secret)
|
|
return api_secret_status or ApiSecretStatus.not_exist
|
|
|
|
# 保存秘钥状态
|
|
async def save_secret(self, secret: str, status: ApiSecretStatus):
|
|
print("save_secret", secret)
|
|
async with redis_utils.get_redis_connection() as redis_client:
|
|
await redis_client.set(f"api_secret_status_${secret}", str(status))
|
|
|
|
# 移除秘钥缓存
|
|
async def remove_secret(self, secret: str):
|
|
await remove_redis_cache(f"api_secret_status_${secret}")
|
|
|
|
|
|
api_secret_utils = ApiSecretUtils()
|