cache.py 1.4 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  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), timeout=10)
  12. @typing.overload
  13. def get(url: str, *, json: typing.Literal[True]=True, headers: httpx._types.HeaderTypes|None=None) -> typing.Any:
  14. ...
  15. @typing.overload
  16. def get(url: str, *, json: typing.Literal[False], headers: httpx._types.HeaderTypes|None=None) -> str:
  17. ...
  18. def get(url: str, *, json=True, headers=None) -> typing.Any:
  19. parsed = urllib.parse.urlparse(url)
  20. assert parsed.hostname is not None
  21. cache_filename = urllib.parse.quote(parsed.path.removeprefix('/'), safe='')
  22. if json:
  23. cache_filename += '.cbor'
  24. cache_filename += '.xz'
  25. cache_path = CACHE_DIR / parsed.hostname / cache_filename
  26. try:
  27. if cache_path.stat().st_mtime > time.time() - 600: # less than 10 minutes old
  28. with lzma.open(cache_path, 'rb') as f:
  29. if json:
  30. return cbor2.load(f)
  31. else:
  32. return f.read().decode('utf-8')
  33. except FileNotFoundError:
  34. pass # fall through
  35. r = client.get(url, headers=headers).raise_for_status()
  36. cache_path.parent.mkdir(parents=True, exist_ok=True)
  37. with lzma.open(cache_path, 'wb') as f:
  38. if json:
  39. data = r.json()
  40. cbor2.dump(data, f)
  41. else:
  42. data = r.text
  43. f.write(data.encode('utf-8'))
  44. return data