"""diskcache 后端:基于 SQLite 的本地持久化缓存,多进程安全。""" from pathlib import Path from typing import Any, Iterator from diskcache import Cache as DiskCache from chongming_cache.backends.base import CacheBackend class DiskBackend(CacheBackend): """包装 diskcache.Cache,提供统一接口。 Args: path: 缓存目录(SQLite 数据文件所在目录),父目录不存在时自动创建。 """ def __init__(self, path: str | Path): self._cache = DiskCache(path) @property def client(self) -> DiskCache: """底层 diskcache.Cache 实例(逃生口)。""" return self._cache def __contains__(self, key) -> bool: return key in self._cache def __getitem__(self, key) -> Any: return self._cache[key] def __setitem__(self, key, value) -> None: self._cache[key] = value def __delitem__(self, key) -> None: del self._cache[key] def __len__(self) -> int: return len(self._cache) def __iter__(self) -> Iterator: return iter(self._cache) def get(self, key, default=None) -> Any: return self._cache.get(key, default) def set(self, key, value, expire: float | None = None) -> None: self._cache.set(key, value, expire=expire) def set_nx(self, key, value, expire: float | None = None) -> bool: # diskcache.add 仅在 key 不存在时写入,返回是否成功(SET NX 语义)且进程安全 return bool(self._cache.add(key, value, expire=expire)) def touch(self, key, expire: float | None = None) -> bool: return self._cache.touch(key, expire=expire) def delete(self, key) -> bool: return self._cache.delete(key) def clear(self) -> None: self._cache.clear() def close(self) -> None: self._cache.close()