(self, key: Hashable, ttl: int)
| 117 | self._lock.release() |
| 118 | |
| 119 | def keep(self, key: Hashable, ttl: int) -> bool: |
| 120 | if not self._lock.acquire(blocking=True, timeout=LOCK_TIMEOUT): |
| 121 | raise Exception("DataCache.keep" + ERR_LOCK_FAILED) |
| 122 | try: |
| 123 | if key in self._base: |
| 124 | _, value = self._base.get(key) |
| 125 | self._base[key] = (self._get_ttl_time(ttl), value) |
| 126 | return True |
| 127 | return False |
| 128 | finally: |
| 129 | self._lock.release() |
| 130 | |
| 131 | def put(self, key: Hashable, value: Any, ttl: int) -> bool: |
| 132 | if not self._lock.acquire(blocking=True, timeout=LOCK_TIMEOUT): |
no test coverage detected