cache.py 957 B

123456789101112131415161718192021222324252627282930313233
  1. import lzma
  2. import pathlib
  3. import time
  4. import typing
  5. import cbor2
  6. import httpx
  7. import urllib.parse
  8. if typing.TYPE_CHECKING:
  9. import httpx._types
  10. CACHE_DIR = pathlib.Path(__file__).parent / 'cache'
  11. client = httpx.Client(transport=httpx.HTTPTransport(http2=True, retries=2))
  12. def get(url: str, *, headers: httpx._types.HeaderTypes|None=None) -> typing.Any:
  13. parsed = urllib.parse.urlparse(url)
  14. assert parsed.hostname is not None
  15. cache_path = CACHE_DIR / parsed.hostname / (urllib.parse.quote(parsed.path.removeprefix('/'), safe='') + '.cbor.xz')
  16. try:
  17. if cache_path.stat().st_mtime > time.time() - 600: # less than 10 minutes old
  18. with lzma.open(cache_path, 'rb') as f:
  19. return cbor2.load(f)
  20. except FileNotFoundError:
  21. pass # fall through
  22. r = client.get(url, headers=headers).raise_for_status()
  23. cache_path.parent.mkdir(parents=True, exist_ok=True)
  24. with lzma.open(cache_path, 'wb') as f:
  25. data = r.json()
  26. cbor2.dump(data, f)
  27. return data