Spaces:
Running
on
A100
Running
on
A100
File size: 3,680 Bytes
2d5e35a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
"""Local cache module to replace Redis
Uses diskcache as backend, provides Redis-compatible API.
Supports persistent storage and TTL expiration.
"""
import json
import os
from typing import Any, Optional
from threading import Lock
try:
from diskcache import Cache
HAS_DISKCACHE = True
except ImportError:
HAS_DISKCACHE = False
class LocalCache:
"""
Local cache implementation with Redis-compatible API.
Uses diskcache as backend, supports persistence and TTL.
"""
_instance = None
_lock = Lock()
def __new__(cls, cache_dir: Optional[str] = None):
"""Singleton pattern"""
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self, cache_dir: Optional[str] = None):
if getattr(self, '_initialized', False):
return
if not HAS_DISKCACHE:
raise ImportError(
"diskcache not installed. Run: pip install diskcache"
)
if cache_dir is None:
cache_dir = os.path.join(
os.path.dirname(os.path.dirname(__file__)),
".cache",
"local_redis"
)
os.makedirs(cache_dir, exist_ok=True)
self._cache = Cache(cache_dir)
self._initialized = True
def set(self, name: str, value: Any, ex: Optional[int] = None) -> bool:
"""
Set key-value pair
Args:
name: Key name
value: Value (auto-serialize dict/list)
ex: Expiration time (seconds)
Returns:
bool: Success status
"""
if isinstance(value, (dict, list)):
value = json.dumps(value, ensure_ascii=False)
self._cache.set(name, value, expire=ex)
return True
def get(self, name: str) -> Optional[str]:
"""Get value"""
return self._cache.get(name)
def delete(self, name: str) -> int:
"""Delete key, returns number of deleted items"""
return 1 if self._cache.delete(name) else 0
def exists(self, name: str) -> bool:
"""Check if key exists"""
return name in self._cache
def keys(self, pattern: str = "*") -> list:
"""
Get list of matching keys
Note: Simplified implementation, only supports prefix and full matching
"""
if pattern == "*":
return list(self._cache.iterkeys())
prefix = pattern.rstrip("*")
return [k for k in self._cache.iterkeys() if k.startswith(prefix)]
def expire(self, name: str, seconds: int) -> bool:
"""Set key expiration time"""
value = self._cache.get(name)
if value is not None:
self._cache.set(name, value, expire=seconds)
return True
return False
def ttl(self, name: str) -> int:
"""
Get remaining time to live (seconds)
Note: diskcache does not directly support TTL queries
"""
if name in self._cache:
return -1 # Exists but TTL unknown
return -2 # Key does not exist
def close(self):
"""Close cache connection"""
if hasattr(self, '_cache'):
self._cache.close()
# Lazily initialized global instance
_local_cache: Optional[LocalCache] = None
def get_local_cache(cache_dir: Optional[str] = None) -> LocalCache:
"""Get local cache instance"""
global _local_cache
if _local_cache is None:
_local_cache = LocalCache(cache_dir)
return _local_cache
|