| 58 | |
| 59 | |
| 60 | class CachePool: |
| 61 | def __init__(self, cache_num: int = -1): |
| 62 | self._cache_num = cache_num |
| 63 | self._cache = OrderedDict() |
| 64 | self.atomic = threading.RLock() |
| 65 | |
| 66 | def keys(self) -> List[str]: |
| 67 | return list(self._cache.keys()) |
| 68 | |
| 69 | def _check_count(self): |
| 70 | if isinstance(self._cache_num, int) and self._cache_num > 0: |
| 71 | while len(self._cache) > self._cache_num: |
| 72 | self._cache.popitem(last=False) |
| 73 | |
| 74 | def get(self, key: str) -> ThreadSafeObject: |
| 75 | if cache := self._cache.get(key): |
| 76 | cache.wait_for_loading() |
| 77 | return cache |
| 78 | |
| 79 | def set(self, key: str, obj: ThreadSafeObject) -> ThreadSafeObject: |
| 80 | self._cache[key] = obj |
| 81 | self._check_count() |
| 82 | return obj |
| 83 | |
| 84 | def pop(self, key: str = None) -> ThreadSafeObject: |
| 85 | if key is None: |
| 86 | return self._cache.popitem(last=False) |
| 87 | else: |
| 88 | return self._cache.pop(key, None) |
| 89 | |
| 90 | def acquire(self, key: Union[str, Tuple], owner: str = "", msg: str = ""): |
| 91 | cache = self.get(key) |
| 92 | if cache is None: |
| 93 | raise RuntimeError(f"请求的资源 {key} 不存在") |
| 94 | elif isinstance(cache, ThreadSafeObject): |
| 95 | self._cache.move_to_end(key) |
| 96 | return cache.acquire(owner=owner, msg=msg) |
| 97 | else: |
| 98 | return cache |
nothing calls this directly
no outgoing calls
no test coverage detected