feat: cached fetch
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
# Project
|
||||
.cached/
|
||||
# ---> Python
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
import json as json_module
|
||||
import os
|
||||
from typing import cast
|
||||
import requests
|
||||
|
||||
CACHE_DIR = ".cached"
|
||||
DEBUG = True
|
||||
|
||||
|
||||
def _escape_path(url: str) -> str:
|
||||
if not os.path.exists(CACHE_DIR):
|
||||
os.makedirs(CACHE_DIR)
|
||||
path = url
|
||||
for char in [":", "/", "?", "&", "=", "%", ".", " "]:
|
||||
path = path.replace(char, "_")
|
||||
return os.path.join(CACHE_DIR, path[:250])
|
||||
|
||||
|
||||
def text(url: str) -> str:
|
||||
cache_path = _escape_path(url)
|
||||
|
||||
if os.path.exists(cache_path):
|
||||
with open(cache_path, "r", encoding="utf-8") as f:
|
||||
if DEBUG:
|
||||
print(f"From cache: {url}")
|
||||
return f.read()
|
||||
|
||||
response = requests.get(url)
|
||||
response.raise_for_status()
|
||||
content = response.text
|
||||
|
||||
with open(cache_path, "w", encoding="utf-8") as f:
|
||||
_ = f.write(content)
|
||||
|
||||
if DEBUG:
|
||||
print(f"Fetch: {url}")
|
||||
return content
|
||||
|
||||
|
||||
type JVal = dict[str, "JVal"] | list["JVal"] | str | int | float | bool | None
|
||||
|
||||
|
||||
def json(url: str) -> JVal:
|
||||
raw_text = text(url)
|
||||
return cast(JVal, json_module.loads(raw_text))
|
||||
Reference in New Issue
Block a user