38 lines
1.5 KiB
Python
38 lines
1.5 KiB
Python
import json
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
USER_AGENT = "FlightMap/0.1 (local non-commercial flight mapping)"
|
|
|
|
|
|
def request_bytes(url: str, *, max_bytes: int = 32 * 1024 * 1024) -> bytes:
|
|
request = Request(url, headers={"User-Agent": USER_AGENT})
|
|
try:
|
|
with urlopen(request, timeout=60) as response:
|
|
content = response.read(max_bytes + 1)
|
|
except (URLError, TimeoutError) as exc:
|
|
raise RuntimeError(f"Raster download failed: {exc}") from exc
|
|
if len(content) > max_bytes:
|
|
raise RuntimeError("Raster download exceeds the 32 MiB limit.")
|
|
return content
|
|
|
|
|
|
def request_json(
|
|
url: str, *, data: bytes | None = None, headers: dict[str, str] | None = None
|
|
) -> Any:
|
|
request = Request(url, data=data, headers={"User-Agent": USER_AGENT, **(headers or {})})
|
|
try:
|
|
with urlopen(request, timeout=60) as response:
|
|
return json.load(response)
|
|
except HTTPError as exc:
|
|
retry = exc.headers.get("Retry-After")
|
|
suffix = f"; retry after {retry}" if retry else ""
|
|
raise RuntimeError(f"Remote service returned HTTP {exc.code}{suffix}.") from exc
|
|
except (URLError, TimeoutError) as exc:
|
|
raise RuntimeError(
|
|
f"Remote service connection failed: {exc.reason if isinstance(exc, URLError) else exc}"
|
|
) from exc
|
|
except (ValueError, UnicodeError) as exc:
|
|
raise RuntimeError("Remote service returned invalid JSON.") from exc
|