from __future__ import annotations from typing import List, Optional try: import redis except Exception: redis = None class CacheBase: """缓存接口""" def get(self, key: str) -> Optional[str]: raise NotImplementedError def set(self, key: str, value: str, ttl: int = None) -> None: raise NotImplementedError def delete(self, key: str) -> None: raise NotImplementedError def keys(self, pattern: str) -> List[str]: raise NotImplementedError class NoopCache(CacheBase): """空实现缓存""" def get(self, key: str) -> Optional[str]: return None def set(self, key: str, value: str, ttl: int = None) -> None: return None def delete(self, key: str) -> None: return None def keys(self, pattern: str) -> List[str]: return [] class RedisCache(CacheBase): """Redis 缓存实现""" def __init__(self, url: str, db: int = 0): if redis is None: raise ImportError("未安装 redis 依赖") self._client = redis.Redis.from_url(url, db=db, decode_responses=True) def get(self, key: str) -> Optional[str]: return self._client.get(key) def set(self, key: str, value: str, ttl: int = None) -> None: if ttl: self._client.set(key, value, ex=ttl) else: self._client.set(key, value) def delete(self, key: str) -> None: self._client.delete(key) def keys(self, pattern: str) -> List[str]: return self._client.keys(pattern)