|
|
@@ -1,3 +1,4 @@
|
|
|
+import datetime
|
|
|
import lzma
|
|
|
import pathlib
|
|
|
import time
|
|
|
@@ -10,16 +11,20 @@ import urllib.parse
|
|
|
if typing.TYPE_CHECKING:
|
|
|
import httpx._types
|
|
|
|
|
|
+ONE_DAY = datetime.timedelta(days=1)
|
|
|
+
|
|
|
CACHE_DIR = pathlib.Path(__file__).parent / 'cache'
|
|
|
client = httpx.Client(transport=httpx.HTTPTransport(http2=True, retries=2), timeout=10)
|
|
|
|
|
|
@typing.overload
|
|
|
-def get(url: str, *, json: typing.Literal[True]=True, headers: httpx._types.HeaderTypes|None=None) -> typing.Any:
|
|
|
+def get(url: str, *, json: typing.Literal[True]=True, headers: httpx._types.HeaderTypes|None=None,
|
|
|
+ expiry: datetime.timedelta=datetime.timedelta(minutes=10)) -> typing.Any:
|
|
|
...
|
|
|
@typing.overload
|
|
|
-def get(url: str, *, json: typing.Literal[False], headers: httpx._types.HeaderTypes|None=None) -> str:
|
|
|
+def get(url: str, *, json: typing.Literal[False], headers: httpx._types.HeaderTypes|None=None,
|
|
|
+ expiry: datetime.timedelta=datetime.timedelta(minutes=10)) -> str:
|
|
|
...
|
|
|
-def get(url: str, *, json=True, headers=None) -> typing.Any:
|
|
|
+def get(url: str, *, json=True, headers=None, expiry=datetime.timedelta(minutes=10)) -> typing.Any:
|
|
|
parsed = urllib.parse.urlparse(url)
|
|
|
assert parsed.hostname is not None
|
|
|
cache_filename = urllib.parse.quote(parsed.path.removeprefix('/'), safe='')
|
|
|
@@ -29,7 +34,7 @@ def get(url: str, *, json=True, headers=None) -> typing.Any:
|
|
|
cache_path = CACHE_DIR / parsed.hostname / cache_filename
|
|
|
|
|
|
try:
|
|
|
- if cache_path.stat().st_mtime > time.time() - 600: # less than 10 minutes old
|
|
|
+ if cache_path.stat().st_mtime > time.time() - expiry.total_seconds(): # less than 10 minutes old
|
|
|
with lzma.open(cache_path, 'rb') as f:
|
|
|
if json:
|
|
|
return cbor2.load(f)
|