daniel156161
73fdeacbfe
Build and Push Docker Container / build-and-push (push) Successful in 2m25s
- Require Python 3.15 for the server and Python 3.14 for the CLI. - Match the server requirement to quart_common's Python 3.15 features. - Replace future annotations with native deferred evaluation. - Defer CLI command dependencies and QR imports on Python 3.15. - Enable argparse suggestions for mistyped CLI commands. - Remove the obsolete tomllib import fallback. - Update my_helpers and quart_common submodule references. - Lock test dependencies and exclude them from Docker installations. - Add regression coverage for lazy imports and deferred annotations. - Document runtime requirements and standalone CLI validation. - Validate 276 tests on Python 3.15 and 18 CLI tests on Python 3.14. BREAKING CHANGE: The server now requires Python 3.15+ and the standalone CLI requires Python 3.14+.
69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
import time
|
|
from collections import OrderedDict
|
|
|
|
class BlobCache:
|
|
"""Process-local, size-bounded LRU for file bytes.
|
|
|
|
Deliberately not AsyncCache: that layer JSON-encodes values (binary would need
|
|
base64) and its memory backend evicts only on read, so blobs could grow without
|
|
bound. Entries are immutable storage blobs, so a stale read is impossible.
|
|
"""
|
|
|
|
def __init__(self, max_total_bytes: int, max_item_bytes: int, ttl: int = 300):
|
|
self._max_total = max(0, int(max_total_bytes))
|
|
self._max_item = max(0, int(max_item_bytes))
|
|
self._ttl = max(1, int(ttl))
|
|
self._entries: OrderedDict[str, tuple[bytes, float]] = OrderedDict()
|
|
self._total = 0
|
|
self.hits = 0
|
|
self.misses = 0
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return self._max_total > 0 and self._max_item > 0
|
|
|
|
@property
|
|
def current_bytes(self) -> int:
|
|
return self._total
|
|
|
|
def _drop(self, key: str) -> None:
|
|
payload = self._entries.pop(key, None)
|
|
if payload is not None:
|
|
self._total -= len(payload[0])
|
|
|
|
def get(self, key: str, now: float | None = None) -> bytes | None:
|
|
entry = self._entries.get(key)
|
|
if entry is None:
|
|
self.misses += 1
|
|
return None
|
|
|
|
data, expires_at = entry
|
|
if (now if now is not None else time.monotonic()) >= expires_at:
|
|
self._drop(key)
|
|
self.misses += 1
|
|
return None
|
|
|
|
self._entries.move_to_end(key)
|
|
self.hits += 1
|
|
return data
|
|
|
|
def set(self, key: str, data: bytes, now: float | None = None) -> bool:
|
|
if not self.enabled or len(data) > self._max_item:
|
|
return False
|
|
|
|
self._drop(key)
|
|
self._entries[key] = (data, (now if now is not None else time.monotonic()) + self._ttl)
|
|
self._total += len(data)
|
|
|
|
while self._total > self._max_total and self._entries:
|
|
oldest = next(iter(self._entries))
|
|
self._drop(oldest)
|
|
return True
|
|
|
|
def invalidate(self, key: str) -> None:
|
|
self._drop(key)
|
|
|
|
def clear(self) -> None:
|
|
self._entries.clear()
|
|
self._total = 0
|