cache.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. import datetime
  2. import lzma
  3. import pathlib
  4. import time
  5. import typing
  6. import cbor2
  7. import httpx
  8. import urllib.parse
  9. if typing.TYPE_CHECKING:
  10. import httpx._types
  11. ONE_DAY = datetime.timedelta(days=1)
  12. CACHE_DIR = pathlib.Path(__file__).parent / 'cache'
  13. client = httpx.Client(transport=httpx.HTTPTransport(http2=True, retries=2), timeout=10)
  14. @typing.overload
  15. def get(url: str, *, json: typing.Literal[True]=True, headers: httpx._types.HeaderTypes|None=None,
  16. expiry: datetime.timedelta=datetime.timedelta(minutes=10)) -> typing.Any:
  17. ...
  18. @typing.overload
  19. def get(url: str, *, json: typing.Literal[False], headers: httpx._types.HeaderTypes|None=None,
  20. expiry: datetime.timedelta=datetime.timedelta(minutes=10)) -> str:
  21. ...
  22. def get(url: str, *, json=True, headers=None, expiry=datetime.timedelta(minutes=10)) -> typing.Any:
  23. parsed = urllib.parse.urlparse(url)
  24. assert parsed.hostname is not None
  25. cache_filename = urllib.parse.quote(parsed.path.removeprefix('/'), safe='')
  26. if json:
  27. cache_filename += '.cbor'
  28. cache_filename += '.xz'
  29. cache_path = CACHE_DIR / parsed.hostname / cache_filename
  30. try:
  31. if cache_path.stat().st_mtime > time.time() - expiry.total_seconds(): # less than 10 minutes old
  32. with lzma.open(cache_path, 'rb') as f:
  33. if json:
  34. return cbor2.load(f)
  35. else:
  36. return f.read().decode('utf-8')
  37. except FileNotFoundError:
  38. pass # fall through
  39. r = client.get(url, headers=headers).raise_for_status()
  40. cache_path.parent.mkdir(parents=True, exist_ok=True)
  41. with lzma.open(cache_path, 'wb') as f:
  42. if json:
  43. data = r.json()
  44. cbor2.dump(data, f)
  45. else:
  46. data = r.text
  47. f.write(data.encode('utf-8'))
  48. return data