mirror of
https://github.com/kuhyx/utils.git
synced 2026-11-11 16:14:07 +01:00
Krzysztof kuhy Rudnicki
78e92a7565
Some checks failed
dependency freshness / dependencies on newest stable (push) Has been cancelled
file length / file-length (push) Has been cancelled
md-naming / markdown naming convention (push) Has been cancelled
palette drift / palette (push) Has been cancelled
shared gate tests / gate unit tests (push) Has been cancelled
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
"""The only module that touches the network.
|
|
|
|
Two things matter here. First, a single 2s per-host reachability probe
|
|
short-circuits every later request, so an offline run fails in seconds instead
|
|
of timing out once per package across ~900 lookups. Second, failures are
|
|
returned, never raised: the gate must degrade to a cached answer rather than
|
|
crash a pre-commit hook the user is forbidden from bypassing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
from dep_freshness._tables import (
|
|
HTTP_ATTEMPTS,
|
|
HTTP_TIMEOUT,
|
|
PROBE_TIMEOUT,
|
|
USER_AGENT,
|
|
)
|
|
|
|
|
|
class Offline(Exception):
|
|
"""The host is unreachable; callers should fall back to the cache."""
|
|
|
|
|
|
_reachable: dict[str, bool] = {}
|
|
_forced_offline = False
|
|
|
|
|
|
def force_offline(value: bool = True) -> None:
|
|
"""Make every request raise `Offline` without touching the network."""
|
|
global _forced_offline
|
|
_forced_offline = value
|
|
|
|
|
|
def reset_probes() -> None:
|
|
"""Forget cached reachability verdicts (tests, and `--refresh`)."""
|
|
_reachable.clear()
|
|
|
|
|
|
def host_reachable(url: str) -> bool:
|
|
"""One cheap TCP probe per host, memoised for the run."""
|
|
if _forced_offline:
|
|
return False
|
|
parts = urlsplit(url)
|
|
host = parts.hostname or ""
|
|
if host in _reachable:
|
|
return _reachable[host]
|
|
port = parts.port or (443 if parts.scheme == "https" else 80)
|
|
try:
|
|
with socket.create_connection((host, port), timeout=PROBE_TIMEOUT):
|
|
_reachable[host] = True
|
|
except OSError:
|
|
_reachable[host] = False
|
|
return _reachable[host]
|
|
|
|
|
|
def get_text(url: str, accept: str | None = None) -> str | None:
|
|
"""Fetch a body as text, or raise `Offline` / return None on 404.
|
|
|
|
Same contract as `get_json`; Maven metadata is XML, not JSON.
|
|
"""
|
|
if not host_reachable(url):
|
|
raise Offline(url)
|
|
headers = {"User-Agent": USER_AGENT}
|
|
if accept:
|
|
headers["Accept"] = accept
|
|
request = Request(url, headers=headers)
|
|
# A slow registry is retried with a longer timeout each time; an HTTP
|
|
# status is an answer and is never retried.
|
|
for attempt in range(1, HTTP_ATTEMPTS + 1):
|
|
try:
|
|
with urlopen(request, timeout=HTTP_TIMEOUT * attempt) as response:
|
|
return response.read().decode("utf-8")
|
|
except HTTPError as exc:
|
|
if exc.code == 404:
|
|
return None
|
|
raise Offline(f"{url}: HTTP {exc.code}") from exc
|
|
except (URLError, TimeoutError, OSError, ValueError) as exc:
|
|
if attempt == HTTP_ATTEMPTS:
|
|
raise Offline(f"{url}: {exc}") from exc
|
|
return None # the loop always returns or raises; this satisfies the type
|
|
|
|
|
|
def get_json(url: str, accept: str | None = None) -> Any:
|
|
"""Fetch and decode JSON, or raise `Offline` / return None on 404.
|
|
|
|
`Offline` means "ask the cache"; None means "the registry answered and
|
|
this package genuinely is not there", which is a real finding rather than
|
|
a degraded run and must not be papered over with a stale cache entry.
|
|
"""
|
|
body = get_text(url, accept)
|
|
if body is None:
|
|
return None
|
|
try:
|
|
return json.loads(body)
|
|
except ValueError as exc:
|
|
raise Offline(f"{url}: {exc}") from exc
|