wehub-resource-sync
126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
# Standard
|
|
from typing import Callable, Generic, Optional, TypeVar
|
|
import threading
|
|
import time
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class TTLListCache(Generic[T]):
|
|
"""
|
|
Generic list cache with TTL (Time-To-Live) support.
|
|
|
|
This cache stores a list of items of type T and automatically refreshes
|
|
when the cache expires.
|
|
"""
|
|
|
|
def __init__(self, timeout_seconds: float = 30.0):
|
|
"""
|
|
Initialize the cache.
|
|
|
|
Args:
|
|
timeout_seconds: Default TTL in seconds for cache entries.
|
|
"""
|
|
self.cache: list[T] = []
|
|
self.cache_time: float = 0.0
|
|
self.cache_lock = threading.Lock()
|
|
self.timeout_seconds = timeout_seconds
|
|
|
|
def get_cached(
|
|
self,
|
|
get_fresh_data: Callable[[], list[T]],
|
|
timeout_override: Optional[float] = None,
|
|
) -> list[T]:
|
|
"""
|
|
Get cached data, refreshing if expired.
|
|
|
|
Args:
|
|
get_fresh_data: Function to get fresh data when cache is expired
|
|
timeout_override: Optional timeout override in seconds.
|
|
If None, use instance timeout_seconds.
|
|
If 0: cache immediately expires, always get fresh data and update cache.
|
|
If > 0: use specified timeout.
|
|
|
|
Returns:
|
|
List of cached items, either from cache or fresh data
|
|
"""
|
|
current_time = time.time()
|
|
|
|
# Fast path: check cache without lock
|
|
if not self.is_expired(timeout_override, current_time):
|
|
return self.cache
|
|
|
|
# Slow path: refresh cache with lock
|
|
with self.cache_lock:
|
|
# Double-check after acquiring lock
|
|
if not self.is_expired(timeout_override, current_time):
|
|
return self.cache
|
|
|
|
# Refresh cache
|
|
self.cache = get_fresh_data()
|
|
self.cache_time = current_time
|
|
return self.cache
|
|
|
|
def clear(self) -> None:
|
|
"""Clear the cache."""
|
|
with self.cache_lock:
|
|
self.cache.clear()
|
|
self.cache_time = 0.0
|
|
|
|
def is_expired(
|
|
self,
|
|
timeout_override: Optional[float] = None,
|
|
current_time: Optional[float] = None,
|
|
) -> bool:
|
|
"""
|
|
Check if the cache is expired without refreshing it.
|
|
|
|
Args:
|
|
timeout_override: Optional timeout override in seconds.
|
|
If None, use instance timeout_seconds.
|
|
If 0: always considered expired.
|
|
If > 0: use specified timeout.
|
|
current_time: Optional current time. If None, use time.time().
|
|
|
|
Returns:
|
|
True if cache is expired or uninitialized, False otherwise
|
|
"""
|
|
if self.cache_time == 0.0:
|
|
return True
|
|
|
|
effective_timeout = (
|
|
timeout_override if timeout_override is not None else self.timeout_seconds
|
|
)
|
|
if current_time is None:
|
|
current_time = time.time()
|
|
|
|
return current_time - self.cache_time >= effective_timeout
|
|
|
|
def get_cache_age(self) -> float:
|
|
"""
|
|
Get the age of the cache in seconds.
|
|
|
|
Returns:
|
|
Age in seconds. Returns float('inf') if cache is uninitialized.
|
|
"""
|
|
if self.cache_time == 0.0:
|
|
return float("inf")
|
|
return time.time() - self.cache_time
|
|
|
|
def __len__(self) -> int:
|
|
"""Get the number of items in cache (without checking expiration)."""
|
|
return len(self.cache)
|
|
|
|
def __repr__(self) -> str:
|
|
"""String representation of the cache."""
|
|
age = self.get_cache_age()
|
|
if age == float("inf"):
|
|
age_str = "uninitialized"
|
|
else:
|
|
age_str = f"{age:.2f}s old"
|
|
return (
|
|
f"TTLListCache(cache_items={len(self)}, age={age_str}, "
|
|
f"timeout={self.timeout_seconds}s)"
|
|
)
|