Simple LRU response cache for relayable static responses.
| 184 | |
| 185 | |
| 186 | class ResponseCache: |
| 187 | """Simple LRU response cache for relayable static responses.""" |
| 188 | |
| 189 | def __init__(self, max_mb: int = 50): |
| 190 | self._store: dict[str, tuple[bytes, float]] = {} |
| 191 | self._size = 0 |
| 192 | self._max = max_mb * 1024 * 1024 |
| 193 | self.hits = 0 |
| 194 | self.misses = 0 |
| 195 | |
| 196 | def get(self, url: str) -> bytes | None: |
| 197 | entry = self._store.get(url) |
| 198 | if not entry: |
| 199 | self.misses += 1 |
| 200 | return None |
| 201 | raw, expires = entry |
| 202 | if time.time() > expires: |
| 203 | self._size -= len(raw) |
| 204 | del self._store[url] |
| 205 | self.misses += 1 |
| 206 | return None |
| 207 | self.hits += 1 |
| 208 | return raw |
| 209 | |
| 210 | def put(self, url: str, raw_response: bytes, ttl: int = 300): |
| 211 | size = len(raw_response) |
| 212 | if size > self._max // 4 or size == 0: |
| 213 | return |
| 214 | while self._size + size > self._max and self._store: |
| 215 | oldest = next(iter(self._store)) |
| 216 | self._size -= len(self._store[oldest][0]) |
| 217 | del self._store[oldest] |
| 218 | if url in self._store: |
| 219 | self._size -= len(self._store[url][0]) |
| 220 | self._store[url] = (raw_response, time.time() + ttl) |
| 221 | self._size += size |
| 222 | |
| 223 | @staticmethod |
| 224 | def parse_ttl(raw_response: bytes, url: str) -> int: |
| 225 | """Determine cache TTL from response headers and URL.""" |
| 226 | hdr_end = raw_response.find(b"\r\n\r\n") |
| 227 | if hdr_end < 0: |
| 228 | return 0 |
| 229 | hdr = raw_response[:hdr_end].decode(errors="replace").lower() |
| 230 | |
| 231 | if b"HTTP/1.1 200" not in raw_response[:20]: |
| 232 | return 0 |
| 233 | # Scope no-store / private checks to the Cache-Control header line so |
| 234 | # URLs like "Location: /api/private/…" or "Server: private-build" |
| 235 | # don't accidentally suppress caching for cacheable responses. |
| 236 | if re.search(r"cache-control:[^\r\n]*\b(?:no-store|private)\b", hdr): |
| 237 | return 0 |
| 238 | if "set-cookie:" in hdr: |
| 239 | return 0 |
| 240 | |
| 241 | max_age_match = re.search(r"max-age=(\d+)", hdr) |
| 242 | if max_age_match: |
| 243 | return min(int(max_age_match.group(1)), CACHE_TTL_MAX) |