(self)
| 75 | return int(time.time()) >= timestamp |
| 76 | |
| 77 | def clean(self) -> None: |
| 78 | if not self._lock.acquire(blocking=True, timeout=LOCK_TIMEOUT): |
| 79 | raise Exception("DataCache.clean" + ERR_LOCK_FAILED) |
| 80 | try: |
| 81 | # Create a list of expired keys to delete |
| 82 | to_delete = [] |
| 83 | for key in self._base: |
| 84 | ttl, _ = self._base[key] |
| 85 | if self._is_expired(ttl): |
| 86 | to_delete.append(key) |
| 87 | # Remove Items |
| 88 | for key in to_delete: |
| 89 | (_, val) = self._base[key] |
| 90 | if isinstance(val, Task): |
| 91 | log.debug(f"Task {key} expired. Data removed.") |
| 92 | elif isinstance(val, SessionState): |
| 93 | log.debug(f"Session {key} expired. Data removed.") |
| 94 | else: |
| 95 | log.debug(f"Key {key} expired. Data removed.") |
| 96 | del self._base[key] |
| 97 | finally: |
| 98 | self._lock.release() |
| 99 | |
| 100 | def clear(self) -> None: |
| 101 | if not self._lock.acquire(blocking=True, timeout=LOCK_TIMEOUT): |
no test coverage detected