GitHub Viewer
"""Shared async query cache with deduplication and explicit invalidation."""
from __future__ import annotations
import asyncio
import time
from collections import OrderedDict
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Hashable
from .runtime import TaskScope
@dataclass(frozen=True)
class QuerySnapshot:
"""Immutable state shared by every subscriber to a query key."""
data: Any = None
loading: bool = False
error: BaseException | None = None
revision: int = 0
class _Entry:
def __init__(self, initial: Any) -> None:
self.snapshot = QuerySnapshot(data=initial, loading=True)
self.listeners: set[Callable[[], None]] = set()
self.fetcher: Callable[[], Awaitable[Any]] | None = None
self.task: Any = None
self.expires = 0.0
self.generation = 0
def publish(self, *, data: Any, loading: bool, error: BaseException | None = None) -> None:
self.snapshot = QuerySnapshot(data, loading, error, self.snapshot.revision + 1)
for listener in tuple(self.listeners):
listener()
class QueryClient:
"""Own shared requests and cached immutable results for one application.
A final unsubscribe cancels pending work. Cached values remain bounded by
capacity. Invalidation cancels older requests before starting replacements.
"""
def __init__(self, *, capacity: int = 128, stale_time: float = 30) -> None:
if capacity < 1 or stale_time < 0:
raise ValueError("Invalid query cache limits")
self.capacity = capacity
self.stale_time = stale_time
self._entries: OrderedDict[Hashable, _Entry] = OrderedDict()
self._scope = TaskScope("queries")
def _entry(self, key: Hashable, initial: Any = None) -> _Entry:
hash(key)
if key not in self._entries:
self._entries[key] = _Entry(initial)
self._entries.move_to_end(key)
for old_key, old in tuple(self._entries.items()):
if len(self._entries) QuerySnapshot:
"""Read a stable snapshot without starting network work."""
return self._entry(key, initial).snapshot
def subscribe(
self, key: Hashable, fetcher: Callable[[], Awaitable[Any]], listener: Callable[[], None]
) -> Callable[[], None]:
"""Subscribe and start one shared request when cached data is stale."""
entry = self._entry(key)
entry.fetcher = fetcher
entry.listeners.add(listener)
if entry.expires None:
entry.listeners.discard(listener)
if not entry.listeners and entry.task is not None and not entry.task.done():
entry.generation += 1
entry.task.cancel()
entry.task = None
entry.publish(data=entry.snapshot.data, loading=False)
return remove
def _fetch(self, entry: _Entry) -> None:
if entry.fetcher is None or entry.task is not None and not entry.task.done():
return
generation = entry.generation
fetcher = entry.fetcher
entry.publish(data=entry.snapshot.data, loading=True)
async def run() -> None:
try:
result = await fetcher()
if generation == entry.generation:
entry.expires = time.monotonic() + self.stale_time
entry.publish(data=result, loading=False)
except asyncio.CancelledError:
raise
except Exception as error:
if generation == entry.generation:
entry.publish(data=entry.snapshot.data, loading=False, error=error)
entry.task = self._scope.create_task(run())
def invalidate(self, key: Hashable | None = None) -> None:
"""Invalidate one key or every query and refresh active subscribers."""
for entry in [self._entry(key)] if key is not None else list(self._entries.values()):
entry.generation += 1
entry.expires = 0
if entry.task is not None:
entry.task.cancel()
entry.task = None
if entry.listeners:
self._fetch(entry)
def set_data(self, key: Hashable, value: Any) -> None:
"""Publish an immutable optimistic result and supersede older requests."""
entry = self._entry(key)
entry.generation += 1
if entry.task is not None:
entry.task.cancel()
entry.task = None
entry.expires = time.monotonic() + self.stale_time
entry.publish(data=value, loading=False)
def close(self) -> None:
"""Cancel shared tasks and release all cached data and listeners."""
self._scope.close()
self._entries.clear()
_default: QueryClient | None = None
def default_client() -> QueryClient:
"""Return the application's default query cache."""
global _default
if _default is None or _default._scope.closed:
_default = QueryClient()
return _default
def _reset_for_tests() -> None:
global _default
if _default is not None:
_default.close()
_default = None